#![allow(clippy::print_stderr)]
use std::collections::{HashMap, VecDeque};
use std::fmt::Debug;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, RwLock as StdRwLock};
use std::time::{Duration, Instant};
use arc_swap::ArcSwap;
use arrow_array::{ArrayRef, BooleanArray, RecordBatch, new_null_array};
use arrow_schema::Schema as ArrowSchema;
use async_trait::async_trait;
use lance_core::datatypes::Schema;
use lance_core::{Error, Result};
use lance_index::mem_wal::ShardManifest;
use lance_index::vector::hnsw::builder::HnswBuildParams;
use lance_io::object_store::{ObjectStore, ObjectStoreParams};
use log::{debug, error, info, warn};
use object_store::path::Path;
use tokio::sync::{RwLock, mpsc};
use tokio::task::JoinHandle;
use tokio::time::{Interval, interval_at};
use tokio_util::sync::CancellationToken;
use tracing::instrument;
use uuid::Uuid;
pub use super::index::{
BTreeIndexConfig, BTreeMemIndex, FtsIndexConfig, HnswIndexConfig, IndexStore, MemIndexConfig,
MemIndexKind, validate_index_configs,
};
pub use super::memtable::CacheConfig;
pub use super::memtable::MemTable;
pub use super::memtable::batch_store::{BatchStore, StoreFull, StoredBatch};
pub use super::memtable::flush::MemTableFlusher;
pub use super::memtable::scanner::MemTableScanner;
pub use super::util::{WatchableOnceCell, WatchableOnceCellReader};
pub use super::wal::{WalEntry, WalEntryData, WalFlushFailure, WalFlushResult, WalFlusher};
use super::memtable::flush::TriggerMemTableFlush;
use super::observer::WalObserver;
use super::scanner::InMemoryMemTableRef;
use super::scanner::SsTableWarmer;
use super::wal::{
BatchDurableWatcher, TriggerIndexApply, TriggerWalFlush, WalAppender, WalFlushSource,
WalOnlyState, WalRetryConfig, WalTailer, WriterCursors, apply_index_range, empty_flush_result,
};
use super::{TOMBSTONE, relax_non_pk_nullability, schema_with_tombstone};
use crate::session::Session;
use super::manifest::ShardManifestStore;
#[derive(Debug, Clone)]
pub struct ShardWriterConfig {
pub shard_id: Uuid,
pub shard_spec_id: u32,
pub durable_write: bool,
pub max_wal_buffer_size: usize,
pub max_wal_flush_interval: Option<Duration>,
pub max_wal_persist_retries: usize,
pub wal_persist_retry_base_delay: Duration,
pub max_memtable_size: usize,
pub max_memtable_rows: usize,
pub max_memtable_batches: usize,
pub manifest_scan_batch_size: usize,
pub max_unflushed_memtable_bytes: usize,
pub backpressure_log_interval: Duration,
pub stats_log_interval: Option<Duration>,
pub frozen_memtable_grace: Duration,
pub enable_memtable: bool,
pub hnsw_params: HashMap<String, HnswBuildParams>,
pub warmer: Option<Arc<dyn SsTableWarmer>>,
pub observer: Option<Arc<dyn WalObserver>>,
pub store_params: Option<ObjectStoreParams>,
pub session: Option<Arc<Session>>,
pub backpressure: Option<Arc<dyn BackpressureController>>,
}
impl Default for ShardWriterConfig {
fn default() -> Self {
Self {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 10 * 1024 * 1024, max_wal_flush_interval: Some(Duration::from_millis(100)), max_wal_persist_retries: 3,
wal_persist_retry_base_delay: Duration::from_millis(50),
max_memtable_size: 256 * 1024 * 1024, max_memtable_rows: 100_000, max_memtable_batches: 8_000, manifest_scan_batch_size: 2,
max_unflushed_memtable_bytes: 1024 * 1024 * 1024, backpressure_log_interval: Duration::from_secs(30),
stats_log_interval: Some(Duration::from_secs(60)), frozen_memtable_grace: Duration::ZERO,
enable_memtable: true,
hnsw_params: HashMap::new(),
warmer: None,
observer: None,
store_params: None,
session: None,
backpressure: None,
}
}
}
impl ShardWriterConfig {
pub fn new(shard_id: Uuid) -> Self {
Self {
shard_id,
..Default::default()
}
}
pub fn with_shard_spec_id(mut self, spec_id: u32) -> Self {
self.shard_spec_id = spec_id;
self
}
pub fn with_durable_write(mut self, durable: bool) -> Self {
self.durable_write = durable;
self
}
pub fn with_max_wal_buffer_size(mut self, size: usize) -> Self {
self.max_wal_buffer_size = size;
self
}
pub fn with_max_wal_flush_interval(mut self, interval: Duration) -> Self {
self.max_wal_flush_interval = Some(interval);
self
}
pub fn with_max_wal_persist_retries(mut self, retries: usize) -> Self {
self.max_wal_persist_retries = retries;
self
}
pub fn with_wal_persist_retry_base_delay(mut self, delay: Duration) -> Self {
self.wal_persist_retry_base_delay = delay;
self
}
pub fn with_max_memtable_size(mut self, size: usize) -> Self {
self.max_memtable_size = size;
self
}
pub fn with_max_memtable_rows(mut self, rows: usize) -> Self {
self.max_memtable_rows = rows;
self
}
pub fn with_max_memtable_batches(mut self, batches: usize) -> Self {
self.max_memtable_batches = batches;
self
}
pub fn with_manifest_scan_batch_size(mut self, size: usize) -> Self {
self.manifest_scan_batch_size = size;
self
}
pub fn with_max_unflushed_memtable_bytes(mut self, size: usize) -> Self {
self.max_unflushed_memtable_bytes = size;
self
}
pub fn with_backpressure(mut self, controller: Arc<dyn BackpressureController>) -> Self {
self.backpressure = Some(controller);
self
}
pub fn with_backpressure_log_interval(mut self, interval: Duration) -> Self {
self.backpressure_log_interval = interval;
self
}
pub fn with_stats_log_interval(mut self, interval: Option<Duration>) -> Self {
self.stats_log_interval = interval;
self
}
pub fn with_frozen_memtable_grace(mut self, grace: Duration) -> Self {
self.frozen_memtable_grace = grace;
self
}
pub fn with_enable_memtable(mut self, enable: bool) -> Self {
self.enable_memtable = enable;
self
}
pub fn with_hnsw_params(
mut self,
index_name: impl Into<String>,
params: HnswBuildParams,
) -> Self {
self.hnsw_params.insert(index_name.into(), params);
self
}
}
type MessageFactory<T> = Box<dyn Fn() -> T + Send + Sync>;
#[async_trait]
pub trait MessageHandler<T: Send + Debug + 'static>: Send {
fn tickers(&mut self) -> Vec<(Duration, MessageFactory<T>)> {
vec![]
}
async fn handle(&mut self, message: T) -> Result<()>;
async fn cleanup(&mut self, _shutdown_ok: bool) -> Result<()> {
Ok(())
}
}
struct TaskDispatcher<T: Send + Debug> {
handler: Box<dyn MessageHandler<T>>,
rx: mpsc::UnboundedReceiver<T>,
cancellation_token: CancellationToken,
name: String,
}
impl<T: Send + Debug + 'static> TaskDispatcher<T> {
async fn run(mut self) -> Result<()> {
let tickers = self.handler.tickers();
let mut ticker_intervals: Vec<(Interval, MessageFactory<T>)> = tickers
.into_iter()
.map(|(duration, factory)| {
let mut interval = interval_at(tokio::time::Instant::now() + duration, duration);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
(interval, factory)
})
.collect();
let result = loop {
if ticker_intervals.is_empty() {
tokio::select! {
biased;
_ = self.cancellation_token.cancelled() => {
debug!("Task '{}' received cancellation", self.name);
break Ok(());
}
msg = self.rx.recv() => {
match msg {
Some(message) => {
if let Err(e) = self.handler.handle(message).await {
error!("Task '{}' error handling message: {}", self.name, e);
}
}
None => {
debug!("Task '{}' channel closed", self.name);
break Ok(());
}
}
}
}
} else {
let first_ticker = ticker_intervals.first_mut().unwrap();
let first_interval = &mut first_ticker.0;
tokio::select! {
biased;
_ = self.cancellation_token.cancelled() => {
debug!("Task '{}' received cancellation", self.name);
break Ok(());
}
msg = self.rx.recv() => {
match msg {
Some(message) => {
if let Err(e) = self.handler.handle(message).await {
error!("Task '{}' error handling message: {}", self.name, e);
}
}
None => {
debug!("Task '{}' channel closed", self.name);
break Ok(());
}
}
}
_ = first_interval.tick() => {
let message = (ticker_intervals[0].1)();
if let Err(e) = self.handler.handle(message).await {
error!("Task '{}' error handling ticker message: {}", self.name, e);
}
}
}
}
};
let cleanup_ok = result.is_ok();
self.handler.cleanup(cleanup_ok).await?;
info!("Task dispatcher '{}' stopped", self.name);
result
}
}
pub struct TaskExecutor {
tasks: StdRwLock<Vec<(String, JoinHandle<Result<()>>)>>,
cancellation_token: CancellationToken,
}
impl TaskExecutor {
pub fn new() -> Self {
Self {
tasks: StdRwLock::new(Vec::new()),
cancellation_token: CancellationToken::new(),
}
}
pub fn add_handler<T: Send + Debug + 'static>(
&self,
name: String,
handler: Box<dyn MessageHandler<T>>,
rx: mpsc::UnboundedReceiver<T>,
) -> Result<()> {
let dispatcher = TaskDispatcher {
handler,
rx,
cancellation_token: self.cancellation_token.clone(),
name: name.clone(),
};
let handle = tokio::spawn(async move { dispatcher.run().await });
self.tasks.write().unwrap().push((name, handle));
Ok(())
}
pub fn add_periodic<F, Fut>(&self, name: String, every: Duration, mut work: F) -> Result<()>
where
F: FnMut() -> Fut + Send + 'static,
Fut: std::future::Future<Output = ()> + Send,
{
let cancellation_token = self.cancellation_token.clone();
let task_name = name.clone();
let handle = tokio::spawn(async move {
let mut interval = interval_at(tokio::time::Instant::now() + every, every);
interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
loop {
tokio::select! {
biased;
_ = cancellation_token.cancelled() => {
debug!("Periodic task '{}' received cancellation", task_name);
return Ok(());
}
_ = interval.tick() => work().await,
}
}
});
self.tasks.write().unwrap().push((name, handle));
Ok(())
}
pub async fn shutdown_all(&self) -> Result<()> {
info!("Shutting down all tasks");
self.cancellation_token.cancel();
let tasks = std::mem::take(&mut *self.tasks.write().unwrap());
let mut first_error = None;
for (name, handle) in tasks {
match handle.await {
Ok(Ok(())) => debug!("Task '{}' completed successfully", name),
Ok(Err(e)) => {
warn!("Task '{}' completed with error: {}", name, e);
if first_error.is_none() {
first_error = Some(e);
}
}
Err(e) => {
error!("Task '{}' panicked: {}", name, e);
if first_error.is_none() {
first_error = Some(Error::internal(format!(
"Task '{name}' panicked during shutdown: {e}"
)));
}
}
}
}
first_error.map_or(Ok(()), Err)
}
}
impl Default for TaskExecutor {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DurabilityResult {
Durable,
Failed(String),
}
impl DurabilityResult {
pub fn ok() -> Self {
Self::Durable
}
pub fn err(msg: impl Into<String>) -> Self {
Self::Failed(msg.into())
}
pub fn is_ok(&self) -> bool {
matches!(self, Self::Durable)
}
pub fn into_result(self) -> Result<()> {
match self {
Self::Durable => Ok(()),
Self::Failed(msg) => Err(Error::io(msg)),
}
}
}
pub type DurabilityWatcher = WatchableOnceCellReader<DurabilityResult>;
pub type DurabilityCell = WatchableOnceCell<DurabilityResult>;
#[derive(Debug, Default)]
pub struct BackpressureStats {
total_count: AtomicU64,
total_wait_ms: AtomicU64,
active_count: AtomicU64,
}
impl BackpressureStats {
pub fn new() -> Self {
Self::default()
}
pub fn record(&self, wait_ms: u64) {
self.total_count.fetch_add(1, Ordering::Relaxed);
self.total_wait_ms.fetch_add(wait_ms, Ordering::Relaxed);
}
pub fn begin_wait(&self) -> BackpressureWaitGuard<'_> {
self.active_count.fetch_add(1, Ordering::Relaxed);
BackpressureWaitGuard(self)
}
pub fn count(&self) -> u64 {
self.total_count.load(Ordering::Relaxed)
}
pub fn total_wait_ms(&self) -> u64 {
self.total_wait_ms.load(Ordering::Relaxed)
}
pub fn snapshot(&self) -> BackpressureStatsSnapshot {
BackpressureStatsSnapshot {
total_count: self.total_count.load(Ordering::Relaxed),
total_wait_ms: self.total_wait_ms.load(Ordering::Relaxed),
active_count: self.active_count.load(Ordering::Relaxed),
}
}
}
#[derive(Debug)]
pub struct BackpressureWaitGuard<'a>(&'a BackpressureStats);
impl Drop for BackpressureWaitGuard<'_> {
fn drop(&mut self) {
self.0.active_count.fetch_sub(1, Ordering::Relaxed);
}
}
#[derive(Debug, Clone, Default)]
pub struct BackpressureStatsSnapshot {
pub total_count: u64,
pub total_wait_ms: u64,
pub active_count: u64,
}
#[derive(Clone)]
pub struct ShardMemory(ShardMemorySource);
#[derive(Clone)]
enum ShardMemorySource {
MemTables(Arc<ArcSwap<ResidentMemTables>>),
Queue(Arc<WalOnlyState>),
#[cfg(test)]
Fake(Arc<dyn Fn() -> usize + Send + Sync>),
}
impl ShardMemory {
fn memtables(tables: Arc<ArcSwap<ResidentMemTables>>) -> Self {
Self(ShardMemorySource::MemTables(tables))
}
fn queue(state: Arc<WalOnlyState>) -> Self {
Self(ShardMemorySource::Queue(state))
}
pub fn active_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => t
.load()
.active
.as_ref()
.map_or(0, InMemoryMemTableRef::resident_bytes),
ShardMemorySource::Queue(q) => q.queue_bytes(),
#[cfg(test)]
ShardMemorySource::Fake(f) => f(),
}
}
pub fn row_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => t
.load()
.active
.as_ref()
.map_or(0, InMemoryMemTableRef::row_bytes),
ShardMemorySource::Queue(q) => q.queue_bytes(),
#[cfg(test)]
ShardMemorySource::Fake(f) => f(),
}
}
pub fn index_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => t
.load()
.active
.as_ref()
.map_or(0, InMemoryMemTableRef::index_bytes),
ShardMemorySource::Queue(_) => 0,
#[cfg(test)]
ShardMemorySource::Fake(_) => 0,
}
}
pub fn frozen_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => t
.load()
.frozen
.iter()
.map(InMemoryMemTableRef::resident_bytes)
.sum(),
ShardMemorySource::Queue(_) => 0,
#[cfg(test)]
ShardMemorySource::Fake(_) => 0,
}
}
pub fn grace_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => t
.load()
.grace
.iter()
.map(InMemoryMemTableRef::resident_bytes)
.sum(),
ShardMemorySource::Queue(_) => 0,
#[cfg(test)]
ShardMemorySource::Fake(_) => 0,
}
}
pub fn unflushed_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => {
let tables = t.load();
tables
.active
.as_ref()
.map_or(0, InMemoryMemTableRef::resident_bytes)
+ tables
.frozen
.iter()
.map(InMemoryMemTableRef::resident_bytes)
.sum::<usize>()
}
ShardMemorySource::Queue(q) => q.queue_bytes(),
#[cfg(test)]
ShardMemorySource::Fake(f) => f(),
}
}
pub fn retained_bytes(&self) -> usize {
match &self.0 {
ShardMemorySource::MemTables(t) => {
let tables = t.load();
tables
.active
.as_ref()
.map_or(0, InMemoryMemTableRef::resident_bytes)
+ tables
.frozen
.iter()
.chain(tables.grace.iter())
.map(InMemoryMemTableRef::resident_bytes)
.sum::<usize>()
}
ShardMemorySource::Queue(q) => q.queue_bytes(),
#[cfg(test)]
ShardMemorySource::Fake(f) => f(),
}
}
pub fn drain(&self) -> Drain {
match &self.0 {
ShardMemorySource::MemTables(t) => {
let tables = t.load();
match tables.oldest_flush.clone() {
Some(flush) => Drain::Flush(flush),
None if !tables.grace.is_empty() => Drain::Background,
None => Drain::Stalled,
}
}
ShardMemorySource::Queue(_) => Drain::Background,
#[cfg(test)]
ShardMemorySource::Fake(_) => Drain::Background,
}
}
}
#[derive(Debug)]
pub enum Drain {
Flush(DurabilityWatcher),
Background,
Stalled,
}
impl Debug for ShardMemory {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ShardMemory")
.field("active_bytes", &self.active_bytes())
.field("frozen_bytes", &self.frozen_bytes())
.field("grace_bytes", &self.grace_bytes())
.finish()
}
}
#[async_trait::async_trait]
pub trait BackpressureController: Send + Sync + Debug {
async fn maybe_apply_backpressure(&self, shard: ShardMemory) -> Result<()>;
fn stats_snapshot(&self) -> BackpressureStatsSnapshot {
BackpressureStatsSnapshot::default()
}
}
fn resolve_backpressure(config: &ShardWriterConfig) -> Arc<dyn BackpressureController> {
match &config.backpressure {
Some(injected) => injected.clone(),
None => Arc::new(LocalBackpressureController::new(config)),
}
}
const DRAIN_POLL_INTERVAL: Duration = Duration::from_millis(10);
const STALL_GRACE: Duration = Duration::from_secs(1);
#[derive(Debug)]
pub struct LocalBackpressureController {
max_unflushed_memtable_bytes: usize,
log_interval: Duration,
stats: Arc<BackpressureStats>,
}
impl LocalBackpressureController {
fn new(config: &ShardWriterConfig) -> Self {
Self {
max_unflushed_memtable_bytes: config.max_unflushed_memtable_bytes,
log_interval: config.backpressure_log_interval,
stats: Arc::new(BackpressureStats::new()),
}
}
pub fn stats(&self) -> &Arc<BackpressureStats> {
&self.stats
}
}
#[async_trait::async_trait]
impl BackpressureController for LocalBackpressureController {
fn stats_snapshot(&self) -> BackpressureStatsSnapshot {
self.stats.snapshot()
}
async fn maybe_apply_backpressure(&self, shard: ShardMemory) -> Result<()> {
let start = std::time::Instant::now();
let mut iteration = 0u32;
let mut active_wait = None;
let mut stalled_since: Option<std::time::Instant> = None;
loop {
let unflushed_memtable_bytes = shard.unflushed_bytes();
if unflushed_memtable_bytes < self.max_unflushed_memtable_bytes {
if iteration > 0 {
let wait_ms = start.elapsed().as_millis() as u64;
self.stats.record(wait_ms);
}
return Ok(());
}
iteration += 1;
if active_wait.is_none() {
active_wait = Some(self.stats.begin_wait());
}
debug!(
"Backpressure triggered: unflushed_bytes={}, max={}, iteration={}",
unflushed_memtable_bytes, self.max_unflushed_memtable_bytes, iteration
);
match shard.drain() {
Drain::Flush(mut mem_watcher) => {
stalled_since = None;
tokio::select! {
_ = mem_watcher.await_value() => {}
_ = tokio::time::sleep(self.log_interval) => {
warn!(
"Backpressure wait timeout, continuing to wait: unflushed_bytes={}, interval={}s, iteration={}",
unflushed_memtable_bytes,
self.log_interval.as_secs(),
iteration
);
}
}
}
Drain::Background => {
stalled_since = None;
tokio::time::sleep(DRAIN_POLL_INTERVAL).await
}
Drain::Stalled => {
let since = *stalled_since.get_or_insert_with(std::time::Instant::now);
if since.elapsed() < STALL_GRACE {
tokio::time::sleep(DRAIN_POLL_INTERVAL).await;
continue;
}
return Err(Error::backpressure(format!(
"shard is at its memtable ceiling with no flush outstanding, so waiting \
cannot drain it: unflushed_bytes={}, max={}, active_bytes={} (of which \
index_bytes={}), frozen_bytes={}. The active memtable seals on resident \
bytes, so reaching here means a flush failed and left its generation \
charged with nothing queued to retry it",
unflushed_memtable_bytes,
self.max_unflushed_memtable_bytes,
shard.active_bytes(),
shard.index_bytes(),
shard.frozen_bytes(),
)));
}
}
}
}
}
#[derive(Debug)]
pub struct WriteResult {
pub batch_positions: std::ops::Range<usize>,
}
struct FrozenMemTable {
memtable: Arc<MemTable>,
flushed_at_ms: Option<u64>,
}
#[derive(Default)]
struct ResidentMemTables {
active: Option<InMemoryMemTableRef>,
frozen: Vec<InMemoryMemTableRef>,
grace: Vec<InMemoryMemTableRef>,
oldest_flush: Option<DurabilityWatcher>,
}
fn publish_memory(memory: &ArcSwap<ResidentMemTables>, state: &WriterState) {
let (grace, frozen) = state
.frozen_memtables
.iter()
.partition::<Vec<_>, _>(|frozen| frozen.flushed_at_ms.is_some());
let refs = |tables: Vec<&FrozenMemTable>| {
tables
.into_iter()
.map(|frozen| in_memory_ref(&frozen.memtable))
.collect()
};
memory.store(Arc::new(ResidentMemTables {
active: Some(in_memory_ref(&state.memtable)),
frozen: refs(frozen),
grace: refs(grace),
oldest_flush: state.frozen_flush_watchers.front().cloned(),
}));
}
struct WriterState {
memtable: MemTable,
last_flushed_wal_entry_position: u64,
frozen_flush_watchers: VecDeque<DurabilityWatcher>,
frozen_memtables: VecDeque<FrozenMemTable>,
flush_requested: bool,
wal_flush_trigger_count: usize,
last_wal_flush_trigger_time: u64,
}
fn in_memory_ref(mt: &MemTable) -> InMemoryMemTableRef {
InMemoryMemTableRef {
batch_store: mt.batch_store(),
index_store: mt
.indexes_arc()
.unwrap_or_else(|| Arc::new(IndexStore::new())),
schema: mt.schema().clone(),
generation: mt.generation(),
}
}
fn start_time() -> std::time::Instant {
use std::sync::OnceLock;
static START: OnceLock<std::time::Instant> = OnceLock::new();
*START.get_or_init(std::time::Instant::now)
}
fn now_millis() -> u64 {
start_time().elapsed().as_millis() as u64
}
struct ReplayResult {
active: MemTable,
next_wal_position: u64,
}
#[allow(clippy::too_many_arguments)]
async fn replay_memtable_from_wal(
object_store: Arc<ObjectStore>,
base_path: Path,
shard_id: Uuid,
our_epoch: u64,
manifest: &ShardManifest,
base_generation: u64,
mut make_memtable: impl FnMut(u64, usize) -> Result<MemTable>,
flusher: &MemTableFlusher,
wal_flusher: &WalFlusher,
index_configs: &[MemIndexConfig],
max_memtable_size: usize,
max_memtable_rows: usize,
max_resident_bytes: usize,
) -> Result<ReplayResult> {
let start_position = manifest.replay_after_wal_entry_position.saturating_add(1);
let tailer = WalTailer::new(object_store, base_path, shard_id);
let mut position = start_position;
let mut active = make_memtable(base_generation, 0)?;
loop {
match tailer.read_entry(position).await? {
None => break,
Some(entry) => {
if entry.writer_epoch > our_epoch {
return Err(Error::fenced_by_peer(format!(
"WAL replay aborted: entry at position {} has writer_epoch {} > our claimed epoch {} for shard {} (writer was fenced during open)",
position, entry.writer_epoch, our_epoch, shard_id
)));
}
if !entry.batches.is_empty() {
let storage_schema = active.schema().clone();
let batches = entry
.batches
.into_iter()
.map(|b| ensure_tombstone_column(b, &storage_schema))
.collect::<Result<Vec<_>>>()?;
let entry_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
if !active.batch_store().is_empty()
&& memtable_reached_flush_threshold(
&active,
max_memtable_size,
max_memtable_rows,
max_resident_bytes,
batches.len(),
entry_rows,
)
{
let store = active.batch_store();
let covered = position.saturating_sub(1);
let generation = active.generation() + 1;
let global_end = store.global_end();
wal_flusher.advance_durable(global_end);
flush_replayed_memtable(
flusher,
&active,
our_epoch,
covered,
global_end,
index_configs,
)
.await?;
active = make_memtable(generation, global_end)?;
}
active.insert_batches_only(batches).await?;
}
position = position.checked_add(1).ok_or_else(|| {
Error::io(format!(
"WAL position overflow during replay for shard {}",
shard_id
))
})?;
}
}
}
if let Some(indexes) = active.indexes_arc() {
let batch_count = active.batch_count();
if batch_count > 0 {
let store = active.batch_store();
let stored: Vec<StoredBatch> = (0..batch_count)
.filter_map(|pos| store.get(pos).cloned())
.collect();
tokio::task::spawn_blocking(move || indexes.insert_batches(&stored))
.await
.map_err(|e| {
Error::internal(format!("WAL replay index update task panicked: {}", e))
})??;
}
}
Ok(ReplayResult {
active,
next_wal_position: position,
})
}
fn memtable_reached_flush_threshold(
memtable: &MemTable,
max_memtable_size: usize,
max_memtable_rows: usize,
max_resident_bytes: usize,
incoming_batches: usize,
incoming_rows: usize,
) -> bool {
let store = memtable.batch_store();
store.row_bytes() >= max_memtable_size
|| memtable_resident_bytes(memtable) >= max_resident_bytes
|| store.remaining_capacity() < incoming_batches
|| store.total_rows().saturating_add(incoming_rows) > max_memtable_rows
}
fn memtable_resident_bytes(memtable: &MemTable) -> usize {
memtable.batch_store().retained_bytes()
+ memtable.indexes().map_or(0, IndexStore::resident_bytes)
+ super::memtable::pk_bloom_filter_bytes()
}
async fn flush_replayed_memtable(
flusher: &MemTableFlusher,
memtable: &MemTable,
epoch: u64,
covered: u64,
durable: usize,
index_configs: &[MemIndexConfig],
) -> Result<()> {
if index_configs.is_empty() {
flusher.flush(memtable, epoch, covered, durable).await?;
} else {
Box::pin(flusher.flush_with_indexes(memtable, epoch, index_configs, covered, durable))
.await?;
}
Ok(())
}
fn pk_index_columns(pk_columns: &[String], pk_field_ids: &[i32]) -> Vec<(String, i32)> {
pk_columns
.iter()
.cloned()
.zip(pk_field_ids.iter().copied())
.collect()
}
fn ensure_tombstone_column(
batch: RecordBatch,
storage_schema: &Arc<ArrowSchema>,
) -> Result<RecordBatch> {
let n = batch.num_rows();
let mut columns: Vec<ArrayRef> = batch.columns().to_vec();
if batch.schema().column_with_name(TOMBSTONE).is_none() {
columns.push(Arc::new(BooleanArray::from(vec![false; n])));
}
RecordBatch::try_new(storage_schema.clone(), columns).map_err(|e| {
Error::invalid_input(format!(
"failed to inject _tombstone column (does the batch match the base table schema?): {}",
e
))
})
}
fn build_tombstone_batch(
keys: &RecordBatch,
storage_schema: &Arc<ArrowSchema>,
pk_columns: &[String],
) -> Result<RecordBatch> {
let n = keys.num_rows();
let mut columns: Vec<ArrayRef> = Vec::with_capacity(storage_schema.fields().len());
for field in storage_schema.fields() {
let name = field.name();
if name == TOMBSTONE {
columns.push(Arc::new(BooleanArray::from(vec![true; n])));
} else if pk_columns.iter().any(|c| c == name) {
let col = keys.column_by_name(name).ok_or_else(|| {
Error::invalid_input(format!(
"delete keys batch is missing primary key column '{}'",
name
))
})?;
columns.push(col.clone());
} else {
columns.push(new_null_array(field.data_type(), n));
}
}
RecordBatch::try_new(storage_schema.clone(), columns).map_err(|e| {
Error::invalid_input(format!(
"failed to build tombstone batch (do the delete keys match the primary key?): {}",
e
))
})
}
struct SharedWriterState {
memory: Arc<ArcSwap<ResidentMemTables>>,
wal_flusher: Arc<WalFlusher>,
wal_flush_tx: mpsc::UnboundedSender<TriggerWalFlush>,
index_apply_tx: mpsc::UnboundedSender<TriggerIndexApply>,
memtable_flush_tx: mpsc::UnboundedSender<TriggerMemTableFlush>,
config: ShardWriterConfig,
schema: Arc<ArrowSchema>,
pk_field_ids: Vec<i32>,
pk_columns: Vec<String>,
max_memtable_batches: usize,
max_memtable_rows: usize,
index_configs: Vec<MemIndexConfig>,
}
impl SharedWriterState {
#[allow(clippy::too_many_arguments)]
fn new(
memory: Arc<ArcSwap<ResidentMemTables>>,
wal_flusher: Arc<WalFlusher>,
wal_flush_tx: mpsc::UnboundedSender<TriggerWalFlush>,
index_apply_tx: mpsc::UnboundedSender<TriggerIndexApply>,
memtable_flush_tx: mpsc::UnboundedSender<TriggerMemTableFlush>,
config: ShardWriterConfig,
schema: Arc<ArrowSchema>,
pk_field_ids: Vec<i32>,
pk_columns: Vec<String>,
max_memtable_batches: usize,
max_memtable_rows: usize,
index_configs: Vec<MemIndexConfig>,
) -> Self {
Self {
memory,
wal_flusher,
wal_flush_tx,
index_apply_tx,
memtable_flush_tx,
config,
schema,
pk_field_ids,
pk_columns,
max_memtable_batches,
max_memtable_rows,
index_configs,
}
}
fn trigger_index_apply(
&self,
batch_store: Arc<BatchStore>,
indexes: Arc<IndexStore>,
end_batch_position: usize,
) -> Result<()> {
self.index_apply_tx
.send(TriggerIndexApply {
batch_store,
indexes,
end_batch_position,
})
.map_err(|_| Error::io("index apply channel closed"))
}
fn freeze_memtable(&self, state: &mut WriterState) -> Result<u64> {
let durable = self.wal_flusher.durable();
let pending_wal_range = state
.memtable
.batch_store()
.pending_wal_flush_range(durable);
let last_wal_entry_position = state.last_flushed_wal_entry_position;
let old_batch_store = state.memtable.batch_store();
let next_generation = state.memtable.generation() + 1;
let next_global_offset = old_batch_store.global_end();
let mut new_memtable = MemTable::with_capacity_at(
self.schema.clone(),
next_generation,
self.pk_field_ids.clone(),
CacheConfig::default(),
self.max_memtable_batches,
next_global_offset,
)?;
let mut indexes = IndexStore::from_configs(
&self.index_configs,
self.max_memtable_rows,
self.max_memtable_batches,
)?;
if !self.pk_columns.is_empty() {
indexes.enable_pk_index(&pk_index_columns(&self.pk_columns, &self.pk_field_ids));
}
indexes.set_durability(Arc::clone(self.wal_flusher.cursors()), next_global_offset);
new_memtable.set_indexes_arc(Arc::new(indexes));
let mut old_memtable = std::mem::replace(&mut state.memtable, new_memtable);
old_memtable.freeze(last_wal_entry_position);
let _memtable_flush_watcher = old_memtable.create_memtable_flush_completion();
let pending_index_apply = match old_memtable.indexes_arc() {
Some(old_indexes) if old_indexes.indexed_count() < old_batch_store.len() => {
Some((old_batch_store.clone(), old_indexes, old_batch_store.len()))
}
_ => None,
};
let pending_wal_flush = if pending_wal_range.is_some() {
let completion_cell: WatchableOnceCell<
std::result::Result<WalFlushResult, WalFlushFailure>,
> = WatchableOnceCell::new();
old_memtable.set_wal_flush_completion(completion_cell.reader());
Some((old_batch_store.len(), completion_cell))
} else {
None
};
let flush_watcher = old_memtable
.get_memtable_flush_watcher()
.expect("Flush watcher should exist after create_memtable_flush_completion");
state.frozen_flush_watchers.push_back(flush_watcher);
let frozen_memtable = Arc::new(old_memtable);
state.frozen_memtables.push_back(FrozenMemTable {
memtable: frozen_memtable.clone(),
flushed_at_ms: None,
});
publish_memory(&self.memory, state);
if let Some((batch_store, indexes, end_batch_position)) = pending_index_apply {
self.trigger_index_apply(batch_store, indexes, end_batch_position)
.inspect_err(|e| self.wal_flusher.poison(e))?;
}
if let Some((end_batch_position, completion_cell)) = pending_wal_flush {
self.wal_flusher
.trigger_flush(
WalFlushSource::BatchStore {
batch_store: old_batch_store,
},
end_batch_position,
Some(completion_cell),
)
.inspect_err(|e| self.wal_flusher.poison(e))?;
}
debug!(
"Frozen memtable generation {}, pending_count = {}",
next_generation - 1,
state.frozen_flush_watchers.len()
);
let _ = self.memtable_flush_tx.send(TriggerMemTableFlush::Flush {
memtable: frozen_memtable,
done: None,
});
Ok(next_generation)
}
fn track_batch_for_wal(
&self,
indexes: Option<Arc<IndexStore>>,
target_indexed: usize,
target_durable: usize,
) -> super::wal::BatchDurableWatcher {
self.wal_flusher
.track_batch(indexes, target_indexed, target_durable)
}
fn maybe_trigger_memtable_flush(
&self,
state: &mut WriterState,
incoming_batches: usize,
incoming_rows: usize,
) -> Result<()> {
if state.flush_requested {
return Ok(());
}
if state.memtable.batch_count() == 0 {
return Ok(());
}
let should_flush = memtable_reached_flush_threshold(
&state.memtable,
self.config.max_memtable_size,
self.config.max_memtable_rows,
self.config.max_unflushed_memtable_bytes,
incoming_batches,
incoming_rows,
);
if should_flush {
state.flush_requested = true;
self.freeze_memtable(state)?;
state.flush_requested = false;
}
Ok(())
}
fn maybe_trigger_wal_flush(&self, state: &mut WriterState) {
let threshold = self.config.max_wal_buffer_size;
let batch_count = state.memtable.batch_count();
let total_bytes = state.memtable.batch_store().row_bytes();
let batch_store = state.memtable.batch_store();
let has_pending = batch_store.pending_wal_flush_count(self.wal_flusher.durable()) > 0;
let time_trigger = if let Some(interval) = self.config.max_wal_flush_interval {
let interval_millis = interval.as_millis() as u64;
let last_trigger = state.last_wal_flush_trigger_time;
let now = now_millis();
if last_trigger == 0 {
state.last_wal_flush_trigger_time = now;
None
} else {
let elapsed = now.saturating_sub(last_trigger);
if elapsed >= interval_millis && has_pending {
state.last_wal_flush_trigger_time = now;
Some(now)
} else {
None
}
}
} else {
None
};
if time_trigger.is_some() {
let _ = self.wal_flush_tx.send(TriggerWalFlush {
source: WalFlushSource::BatchStore { batch_store },
end_batch_position: batch_count,
done: None,
});
return;
}
if threshold == 0 {
return;
}
let thresholds_crossed = total_bytes / threshold;
while state.wal_flush_trigger_count < thresholds_crossed {
state.wal_flush_trigger_count += 1;
state.last_wal_flush_trigger_time = now_millis();
let _ = self.wal_flush_tx.send(TriggerWalFlush {
source: WalFlushSource::BatchStore {
batch_store: batch_store.clone(),
},
end_batch_position: batch_count,
done: None,
});
}
}
}
#[derive(Debug, Default)]
struct WalOnlyTriggerState {
last_trigger_pending_bytes: usize,
last_wal_flush_trigger_time: u64,
}
enum WriterMode {
MemTable {
state: Arc<RwLock<WriterState>>,
writer_state: Arc<SharedWriterState>,
backpressure: Arc<dyn BackpressureController>,
},
WalOnly {
state: Arc<WalOnlyState>,
wal_flush_tx: mpsc::UnboundedSender<TriggerWalFlush>,
trigger: StdRwLock<WalOnlyTriggerState>,
backpressure: Arc<dyn BackpressureController>,
},
}
pub struct ShardWriter {
config: ShardWriterConfig,
epoch: u64,
wal_flusher: Arc<WalFlusher>,
task_executor: Arc<TaskExecutor>,
manifest_store: Arc<ShardManifestStore>,
stats: SharedWriteStats,
mode: WriterMode,
logical_schema: Arc<ArrowSchema>,
}
impl ShardWriter {
#[instrument(name = "sw_open", level = "info", skip_all, fields(shard_id = %config.shard_id, index_count = index_configs.len()))]
pub async fn open(
object_store: Arc<ObjectStore>,
base_path: Path,
base_uri: impl Into<String>,
config: ShardWriterConfig,
schema: Arc<ArrowSchema>,
index_configs: Vec<MemIndexConfig>,
) -> Result<Self> {
if !config.enable_memtable && !index_configs.is_empty() {
return Err(Error::invalid_input(
"indexes require enable_memtable = true; \
WAL-only mode does not maintain in-memory indexes",
));
}
if config.durable_write && config.max_wal_flush_interval.is_none_or(|d| d.is_zero()) {
return Err(Error::invalid_input(
"durable_write requires a positive max_wal_flush_interval: with no \
flush ticker a durable put has nothing to drive its WAL append and \
would block until close",
));
}
let logical_schema = schema;
let tombstoned = schema_with_tombstone(&logical_schema);
let base_uri = base_uri.into();
let shard_id = config.shard_id;
let manifest_store = Arc::new(ShardManifestStore::new(
object_store.clone(),
&base_path,
shard_id,
config.manifest_scan_batch_size,
));
let memtable_validation = if config.enable_memtable {
let lance_schema = Schema::try_from(tombstoned.as_ref())?;
let pk_fields = lance_schema.unenforced_primary_key();
let pk_field_ids: Vec<i32> = pk_fields.iter().map(|f| f.id).collect();
let pk_columns: Vec<String> = pk_fields.iter().map(|f| f.name.clone()).collect();
validate_index_configs(
&index_configs,
tombstoned.as_ref(),
&lance_schema,
&pk_columns,
)?;
if config.backpressure.is_none() {
if config.max_memtable_size == 0 {
return Err(Error::invalid_input(
"max_memtable_size must be greater than zero: it is both the \
seal threshold for row data and the headroom reserved for rows \
under max_unflushed_memtable_bytes, and at zero a writer with \
in-memory indexes can be at its ceiling before its first row",
));
}
let mut indexes = IndexStore::from_configs(
&index_configs,
config.max_memtable_rows,
config.max_memtable_batches,
)?;
if !pk_columns.is_empty() {
indexes.enable_pk_index(&pk_index_columns(&pk_columns, &pk_field_ids));
}
let reserved = indexes.resident_bytes() + super::memtable::pk_bloom_filter_bytes();
let needed = reserved.saturating_add(config.max_memtable_size);
if needed > config.max_unflushed_memtable_bytes {
return Err(Error::invalid_input(format!(
"in-memory indexes reserve {reserved} bytes at \
max_memtable_rows={}, and max_memtable_size={} must fit alongside them, \
needing {needed} bytes; max_unflushed_memtable_bytes={} is below that, \
so the active memtable would cross the backpressure ceiling before \
accruing enough row bytes to seal, stalling every write. Raise \
max_unflushed_memtable_bytes to at least {needed}, or lower \
max_memtable_rows / max_memtable_size",
config.max_memtable_rows,
config.max_memtable_size,
config.max_unflushed_memtable_bytes,
)));
}
}
let storage_schema = relax_non_pk_nullability(&tombstoned, &pk_columns);
Some((pk_field_ids, pk_columns, storage_schema))
} else {
None
};
let (epoch, manifest) = manifest_store.claim_epoch(config.shard_spec_id).await?;
info!(
"Opened ShardWriter for shard {} (epoch {}, generation {}, enable_memtable {})",
shard_id, epoch, manifest.current_generation, config.enable_memtable
);
let position_hint_seed = manifest
.wal_entry_position_last_seen
.max(manifest.replay_after_wal_entry_position)
.saturating_add(1);
let wal_appender = Arc::new(WalAppender::with_claimed_epoch(
object_store.clone(),
base_path.clone(),
shard_id,
manifest_store.clone(),
epoch,
position_hint_seed,
WalRetryConfig {
max_retries: config.max_wal_persist_retries,
base_delay: config.wal_persist_retry_base_delay,
},
));
if epoch >= 2 {
wal_appender.write_fence_sentinel().await?;
}
let cursors = Arc::new(WriterCursors::new(config.durable_write));
let mut wal_flusher = WalFlusher::with_cursors(wal_appender, cursors);
let (wal_flush_tx, wal_flush_rx) = mpsc::unbounded_channel();
wal_flusher.set_flush_channel(wal_flush_tx.clone());
let wal_flusher = Arc::new(wal_flusher);
let stats = new_shared_stats();
let task_executor = Arc::new(TaskExecutor::new());
let mode = if config.enable_memtable {
let (pk_field_ids, pk_columns, storage_schema) = memtable_validation
.expect("memtable_validation is Some when enable_memtable is true");
Self::open_memtable_mode(
&config,
&storage_schema,
&manifest,
&index_configs,
pk_field_ids,
pk_columns,
wal_flusher.clone(),
wal_flush_tx,
wal_flush_rx,
object_store.clone(),
base_path,
base_uri,
shard_id,
epoch,
manifest_store.clone(),
stats.clone(),
&task_executor,
)
.await?
} else {
Self::open_wal_only_mode(
&config,
wal_flusher.clone(),
wal_flush_tx,
wal_flush_rx,
stats.clone(),
&task_executor,
)?
};
Ok(Self {
config,
epoch,
wal_flusher,
task_executor,
manifest_store,
stats,
mode,
logical_schema,
})
}
#[allow(clippy::too_many_arguments)]
async fn open_memtable_mode(
config: &ShardWriterConfig,
schema: &Arc<ArrowSchema>,
manifest: &ShardManifest,
index_configs: &[MemIndexConfig],
pk_field_ids: Vec<i32>,
pk_columns: Vec<String>,
wal_flusher: Arc<WalFlusher>,
wal_flush_tx: mpsc::UnboundedSender<TriggerWalFlush>,
wal_flush_rx: mpsc::UnboundedReceiver<TriggerWalFlush>,
object_store: Arc<ObjectStore>,
base_path: Path,
base_uri: String,
shard_id: Uuid,
epoch: u64,
manifest_store: Arc<ShardManifestStore>,
stats: SharedWriteStats,
task_executor: &Arc<TaskExecutor>,
) -> Result<WriterMode> {
let make_bound_memtable = |generation: u64, global_offset: usize| -> Result<MemTable> {
let mut memtable = MemTable::with_capacity_at(
schema.clone(),
generation,
pk_field_ids.clone(),
CacheConfig::default(),
config.max_memtable_batches,
global_offset,
)?;
let mut indexes = IndexStore::from_configs(
index_configs,
config.max_memtable_rows,
config.max_memtable_batches,
)?;
if !pk_columns.is_empty() {
indexes.enable_pk_index(&pk_index_columns(&pk_columns, &pk_field_ids));
}
indexes.set_durability(Arc::clone(wal_flusher.cursors()), global_offset);
memtable.set_indexes_arc(Arc::new(indexes));
Ok(memtable)
};
let flusher = Arc::new(
MemTableFlusher::new(
object_store.clone(),
base_path.clone(),
base_uri.clone(),
shard_id,
manifest_store.clone(),
)
.with_warmer(config.warmer.clone())
.with_storage_context(config.store_params.clone(), config.session.clone()),
);
let ReplayResult {
active: memtable,
next_wal_position,
} = replay_memtable_from_wal(
object_store.clone(),
base_path.clone(),
shard_id,
epoch,
manifest,
manifest.current_generation,
make_bound_memtable,
&flusher,
&wal_flusher,
index_configs,
config.max_memtable_size,
config.max_memtable_rows,
config.max_unflushed_memtable_bytes,
)
.await?;
let replayed_through = next_wal_position.saturating_sub(1);
if replayed_through > manifest.wal_entry_position_last_seen
&& let Err(error) = manifest_store
.commit_update(epoch, |current| ShardManifest {
version: current.next_version(),
wal_entry_position_last_seen: current
.wal_entry_position_last_seen
.max(replayed_through),
..current.clone()
})
.await
{
warn!(
"failed to publish WAL read cursor {} for shard {}: {}",
replayed_through, shard_id, error
);
}
wal_flusher.advance_durable(memtable.batch_store().global_end());
wal_flusher
.wal_appender()
.seed_next_position(next_wal_position)
.await;
let initial_covered_wal_entry_position = next_wal_position.saturating_sub(1);
let memory = Arc::new(ArcSwap::<ResidentMemTables>::default());
let state = WriterState {
memtable,
last_flushed_wal_entry_position: initial_covered_wal_entry_position,
frozen_flush_watchers: VecDeque::new(),
frozen_memtables: VecDeque::new(),
flush_requested: false,
wal_flush_trigger_count: 0,
last_wal_flush_trigger_time: 0,
};
publish_memory(&memory, &state);
let state = Arc::new(RwLock::new(state));
let (memtable_flush_tx, memtable_flush_rx) = mpsc::unbounded_channel();
let flusher = Arc::new(
MemTableFlusher::new(object_store, base_path, base_uri, shard_id, manifest_store)
.with_warmer(config.warmer.clone())
.with_storage_context(config.store_params.clone(), config.session.clone()),
);
let wal_handler = WalFlushHandler::new(
wal_flusher.clone(),
Some(state.clone()),
None,
config.max_wal_flush_interval,
stats.clone(),
config.observer.clone(),
);
task_executor.add_handler(
"wal_flusher".to_string(),
Box::new(wal_handler),
wal_flush_rx,
)?;
let memtable_handler = MemTableFlushHandler::new(
state.clone(),
memory.clone(),
flusher,
wal_flusher.clone(),
epoch,
index_configs.to_vec(),
stats.clone(),
config.observer.clone(),
config.frozen_memtable_grace,
);
task_executor.add_handler(
"memtable_flusher".to_string(),
Box::new(memtable_handler),
memtable_flush_rx,
)?;
if !config.frozen_memtable_grace.is_zero() {
let tick = (config.frozen_memtable_grace / 3).max(Duration::from_millis(100));
let sweep_state = state.clone();
let sweep_memory = memory.clone();
let grace = config.frozen_memtable_grace;
task_executor.add_periodic("memtable_grace_sweeper".to_string(), tick, move || {
let state = sweep_state.clone();
let memory = sweep_memory.clone();
async move { sweep_expired_frozen(&state, &memory, grace).await }
})?;
}
let (index_apply_tx, index_apply_rx) = mpsc::unbounded_channel();
let index_handler = IndexApplyHandler {
cursors: Arc::clone(wal_flusher.cursors()),
wal_flusher: wal_flusher.clone(),
stats,
};
task_executor.add_handler(
"index_applier".to_string(),
Box::new(index_handler),
index_apply_rx,
)?;
let writer_state = Arc::new(SharedWriterState::new(
memory,
wal_flusher,
wal_flush_tx,
index_apply_tx,
memtable_flush_tx,
config.clone(),
schema.clone(),
pk_field_ids,
pk_columns,
config.max_memtable_batches,
config.max_memtable_rows,
index_configs.to_vec(),
));
let backpressure = resolve_backpressure(config);
Ok(WriterMode::MemTable {
state,
writer_state,
backpressure,
})
}
fn open_wal_only_mode(
config: &ShardWriterConfig,
wal_flusher: Arc<WalFlusher>,
wal_flush_tx: mpsc::UnboundedSender<TriggerWalFlush>,
wal_flush_rx: mpsc::UnboundedReceiver<TriggerWalFlush>,
stats: SharedWriteStats,
task_executor: &Arc<TaskExecutor>,
) -> Result<WriterMode> {
let state = Arc::new(WalOnlyState::default());
let wal_handler = WalFlushHandler::new(
wal_flusher,
None,
Some(state.clone()),
config.max_wal_flush_interval,
stats,
config.observer.clone(),
);
task_executor.add_handler(
"wal_flusher".to_string(),
Box::new(wal_handler),
wal_flush_rx,
)?;
let backpressure = resolve_backpressure(config);
Ok(WriterMode::WalOnly {
state,
wal_flush_tx,
trigger: StdRwLock::new(WalOnlyTriggerState::default()),
backpressure,
})
}
#[instrument(name = "sw_put", level = "info", skip_all, fields(batch_count = batches.len(), shard_id = %self.config.shard_id))]
pub async fn put(&self, batches: Vec<RecordBatch>) -> Result<WriteResult> {
Self::validate_non_empty(&batches)?;
self.validate_against_logical_schema(&batches)?;
match &self.mode {
WriterMode::MemTable {
state,
writer_state,
backpressure,
} => {
let batches = batches
.into_iter()
.map(|b| ensure_tombstone_column(b, &writer_state.schema))
.collect::<Result<Vec<_>>>()?;
self.put_memtable(batches, state, writer_state, backpressure)
.await
}
WriterMode::WalOnly {
state,
wal_flush_tx,
trigger,
backpressure,
} => {
self.put_wal_only(batches, state, wal_flush_tx, trigger, backpressure)
.await
}
}
}
#[instrument(name = "sw_delete", level = "info", skip_all, fields(batch_count = keys.len(), shard_id = %self.config.shard_id))]
pub async fn delete(&self, keys: Vec<RecordBatch>) -> Result<WriteResult> {
let (result, watcher) = self.delete_no_wait(keys).await?;
if let Some(mut watcher) = watcher {
watcher.wait().await?;
}
Ok(result)
}
#[instrument(name = "sw_delete_no_wait", level = "info", skip_all, fields(batch_count = keys.len(), shard_id = %self.config.shard_id))]
pub async fn delete_no_wait(
&self,
keys: Vec<RecordBatch>,
) -> Result<(WriteResult, Option<BatchDurableWatcher>)> {
if keys.is_empty() {
return Err(Error::invalid_input("Cannot delete with empty key list"));
}
for (i, batch) in keys.iter().enumerate() {
if batch.num_rows() == 0 {
return Err(Error::invalid_input(format!("Key batch {} is empty", i)));
}
}
match &self.mode {
WriterMode::MemTable {
state,
writer_state,
backpressure,
} => {
if writer_state.pk_columns.is_empty() {
return Err(Error::invalid_input(
"delete requires a primary key, but this shard has no primary key columns",
));
}
let tombstones = keys
.into_iter()
.map(|k| {
build_tombstone_batch(&k, &writer_state.schema, &writer_state.pk_columns)
})
.collect::<Result<Vec<_>>>()?;
self.put_memtable_no_wait(tombstones, state, writer_state, backpressure)
.await
}
WriterMode::WalOnly { .. } => Err(Error::invalid_input(
"delete is only supported in memtable mode (enable_memtable = true)",
)),
}
}
#[instrument(name = "sw_put_no_wait", level = "info", skip_all, fields(batch_count = batches.len(), shard_id = %self.config.shard_id))]
pub async fn put_no_wait(
&self,
batches: Vec<RecordBatch>,
) -> Result<(WriteResult, Option<BatchDurableWatcher>)> {
Self::validate_non_empty(&batches)?;
self.validate_against_logical_schema(&batches)?;
match &self.mode {
WriterMode::MemTable {
state,
writer_state,
backpressure,
} => {
let batches = batches
.into_iter()
.map(|b| ensure_tombstone_column(b, &writer_state.schema))
.collect::<Result<Vec<_>>>()?;
self.put_memtable_no_wait(batches, state, writer_state, backpressure)
.await
}
WriterMode::WalOnly { .. } => Err(Error::invalid_input(
"put_no_wait is only supported in MemTable mode",
)),
}
}
fn validate_against_logical_schema(&self, batches: &[RecordBatch]) -> Result<()> {
for (i, batch) in batches.iter().enumerate() {
for (col, (expected, actual)) in self
.logical_schema
.fields()
.iter()
.zip(batch.schema().fields())
.enumerate()
{
if expected.name() != actual.name() {
return Err(Error::invalid_input(format!(
"batch {i} column {col} is named '{}', but the base table schema \
declares '{}' at that position",
actual.name(),
expected.name()
)));
}
}
RecordBatch::try_new(self.logical_schema.clone(), batch.columns().to_vec()).map_err(
|e| {
Error::invalid_input(format!(
"batch {i} does not match the base table schema: {e}"
))
},
)?;
}
Ok(())
}
fn validate_non_empty(batches: &[RecordBatch]) -> Result<()> {
if batches.is_empty() {
return Err(Error::invalid_input("Cannot write empty batch list"));
}
for (i, batch) in batches.iter().enumerate() {
if batch.num_rows() == 0 {
return Err(Error::invalid_input(format!("Batch {} is empty", i)));
}
}
Ok(())
}
async fn put_memtable(
&self,
batches: Vec<RecordBatch>,
state_lock: &Arc<RwLock<WriterState>>,
writer_state: &Arc<SharedWriterState>,
backpressure: &Arc<dyn BackpressureController>,
) -> Result<WriteResult> {
let (result, watcher) = self
.put_memtable_no_wait(batches, state_lock, writer_state, backpressure)
.await?;
if let Some(mut watcher) = watcher {
watcher.wait().await?;
}
Ok(result)
}
async fn put_memtable_no_wait(
&self,
batches: Vec<RecordBatch>,
state_lock: &Arc<RwLock<WriterState>>,
writer_state: &Arc<SharedWriterState>,
backpressure: &Arc<dyn BackpressureController>,
) -> Result<(WriteResult, Option<BatchDurableWatcher>)> {
self.wal_flusher.check_poisoned()?;
let incoming_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
if incoming_rows > self.config.max_memtable_rows {
return Err(Error::invalid_input(format!(
"write of {incoming_rows} rows across {} batches exceeds \
max_memtable_rows={}: a write is never split across memtables, and the \
in-memory indexes are sized to that cap. Split the write, or raise \
max_memtable_rows",
batches.len(),
self.config.max_memtable_rows,
)));
}
if ShardMemory::memtables(writer_state.memory.clone()).unflushed_bytes()
>= self.config.max_unflushed_memtable_bytes
{
let mut state = state_lock.write().await;
writer_state.maybe_trigger_memtable_flush(&mut state, 1, 1)?;
}
backpressure
.maybe_apply_backpressure(ShardMemory::memtables(writer_state.memory.clone()))
.await?;
let start = std::time::Instant::now();
let (batch_positions, durable_watcher, batch_store, indexes) = {
let mut state = state_lock.write().await;
writer_state.maybe_trigger_memtable_flush(&mut state, batches.len(), incoming_rows)?;
let results = state.memtable.insert_batches_only(batches).await?;
let batch_store = state.memtable.batch_store();
let indexes = state.memtable.indexes_arc();
let start_pos = results.first().map(|(pos, _, _)| *pos).unwrap_or(0);
let end_pos = results.last().map(|(pos, _, _)| pos + 1).unwrap_or(0);
let batch_positions = start_pos..end_pos;
let durable_watcher = writer_state.track_batch_for_wal(
indexes.clone(),
end_pos,
batch_store.global_offset() + end_pos,
);
writer_state.maybe_trigger_wal_flush(&mut state);
if let Err(e) = writer_state.maybe_trigger_memtable_flush(&mut state, 1, 1) {
warn!("Failed to trigger memtable flush: {}", e);
}
(batch_positions, durable_watcher, batch_store, indexes)
};
self.stats.record_put(start.elapsed());
if let Some(indexes) = indexes {
writer_state.trigger_index_apply(batch_store, indexes, batch_positions.end)?;
}
let watcher = Some(durable_watcher);
Ok((WriteResult { batch_positions }, watcher))
}
async fn put_wal_only(
&self,
batches: Vec<RecordBatch>,
state: &Arc<WalOnlyState>,
wal_flush_tx: &mpsc::UnboundedSender<TriggerWalFlush>,
trigger: &StdRwLock<WalOnlyTriggerState>,
backpressure: &Arc<dyn BackpressureController>,
) -> Result<WriteResult> {
self.wal_flusher.check_poisoned()?;
backpressure
.maybe_apply_backpressure(ShardMemory::queue(state.clone()))
.await?;
let start = std::time::Instant::now();
let batch_positions = state.push(batches);
let durable_watcher = self
.config
.durable_write
.then(|| self.wal_flusher.track_batch(None, 0, batch_positions.end));
self.maybe_trigger_wal_flush_wal_only(
state,
wal_flush_tx,
trigger,
batch_positions.end,
state.queue_bytes(),
);
self.stats.record_put(start.elapsed());
if let Some(mut watcher) = durable_watcher {
watcher.wait().await?;
}
Ok(WriteResult { batch_positions })
}
fn maybe_trigger_wal_flush_wal_only(
&self,
state: &Arc<WalOnlyState>,
wal_flush_tx: &mpsc::UnboundedSender<TriggerWalFlush>,
trigger: &StdRwLock<WalOnlyTriggerState>,
end_batch_position: usize,
pending_bytes: usize,
) {
let threshold = self.config.max_wal_buffer_size;
let has_pending = state.batch_count() > 0;
let mut t = trigger.write().unwrap();
if let Some(interval) = self.config.max_wal_flush_interval {
let interval_millis = interval.as_millis() as u64;
let now = now_millis();
if t.last_wal_flush_trigger_time == 0 {
t.last_wal_flush_trigger_time = now;
} else {
let elapsed = now.saturating_sub(t.last_wal_flush_trigger_time);
if elapsed >= interval_millis && has_pending {
t.last_wal_flush_trigger_time = now;
let _ = wal_flush_tx.send(TriggerWalFlush {
source: WalFlushSource::WalOnly {
state: state.clone(),
},
end_batch_position,
done: None,
});
return;
}
}
}
if threshold == 0 {
return;
}
if pending_bytes < t.last_trigger_pending_bytes {
t.last_trigger_pending_bytes = 0;
}
while pending_bytes >= t.last_trigger_pending_bytes + threshold {
t.last_trigger_pending_bytes += threshold;
t.last_wal_flush_trigger_time = now_millis();
let _ = wal_flush_tx.send(TriggerWalFlush {
source: WalFlushSource::WalOnly {
state: state.clone(),
},
end_batch_position,
done: None,
});
}
}
pub fn stats(&self) -> WriteStatsSnapshot {
self.stats.snapshot()
}
pub fn stats_handle(&self) -> SharedWriteStats {
self.stats.clone()
}
pub fn memory(&self) -> ShardMemory {
match &self.mode {
WriterMode::MemTable { writer_state, .. } => {
ShardMemory::memtables(writer_state.memory.clone())
}
WriterMode::WalOnly { state, .. } => ShardMemory::queue(state.clone()),
}
}
pub async fn manifest(&self) -> Result<Option<ShardManifest>> {
self.manifest_store.latest().await
}
pub fn manifest_store(&self) -> Arc<ShardManifestStore> {
self.manifest_store.clone()
}
pub fn epoch(&self) -> u64 {
self.epoch
}
pub fn shard_id(&self) -> Uuid {
self.config.shard_id
}
pub async fn check_fenced(&self) -> Result<()> {
self.manifest_store.check_fenced(self.epoch).await
}
pub async fn memtable_stats(&self) -> Result<MemTableStats> {
let state_lock = self.memtable_state_lock()?;
let state = state_lock.read().await;
let batch_store = state.memtable.batch_store();
let durable = self.wal_flusher.durable();
let pending_wal = batch_store.pending_wal_flush_stats(durable);
Ok(MemTableStats {
row_count: state.memtable.row_count(),
batch_count: state.memtable.batch_count(),
generation: state.memtable.generation(),
max_buffered_batch_position: batch_store.max_buffered_batch_position(),
durable_batch_count: durable,
global_offset: batch_store.global_offset(),
pending_wal_start_batch_position: pending_wal.start_batch_position,
pending_wal_end_batch_position: pending_wal.end_batch_position,
pending_wal_batch_count: pending_wal.batch_count,
pending_wal_row_count: pending_wal.row_count,
pending_wal_estimated_bytes: pending_wal.estimated_bytes,
frozen_count: state.frozen_memtables.len(),
})
}
pub fn backpressure_stats(&self) -> BackpressureStatsSnapshot {
match &self.mode {
WriterMode::MemTable { backpressure, .. }
| WriterMode::WalOnly { backpressure, .. } => backpressure.stats_snapshot(),
}
}
pub async fn scan(&self) -> Result<MemTableScanner> {
self.wal_flusher.check_poisoned()?;
let state_lock = self.memtable_state_lock()?;
let state = state_lock.read().await;
Ok(state.memtable.scan())
}
pub async fn active_memtable_ref(
&self,
) -> Result<crate::dataset::mem_wal::scanner::InMemoryMemTableRef> {
self.wal_flusher.check_poisoned()?;
let state_lock = self.memtable_state_lock()?;
let state = state_lock.read().await;
Ok(in_memory_ref(&state.memtable))
}
pub async fn in_memory_memtable_refs(
&self,
) -> Result<crate::dataset::mem_wal::scanner::InMemoryMemTables> {
self.wal_flusher.check_poisoned()?;
let state_lock = self.memtable_state_lock()?;
let state = state_lock.read().await;
Ok(crate::dataset::mem_wal::scanner::InMemoryMemTables {
active: in_memory_ref(&state.memtable),
frozen: state
.frozen_memtables
.iter()
.map(|m| in_memory_ref(&m.memtable))
.collect(),
})
}
fn memtable_state_lock(&self) -> Result<&Arc<RwLock<WriterState>>> {
match &self.mode {
WriterMode::MemTable { state, .. } => Ok(state),
WriterMode::WalOnly { .. } => Err(Error::invalid_input(
"MemTable accessor not available when enable_memtable = false (WAL-only mode)",
)),
}
}
pub fn wal_stats(&self) -> WalStats {
WalStats {
next_wal_entry_position: self.wal_flusher.next_wal_entry_position(),
}
}
#[instrument(name = "sw_force_seal_active", level = "info", skip_all, fields(shard_id = %self.config.shard_id, epoch = self.epoch))]
pub async fn force_seal_active(&self) -> Result<SealFence> {
match &self.mode {
WriterMode::MemTable {
state,
writer_state,
..
} => {
self.check_fenced().await?;
self.wal_flusher.check_poisoned()?;
let mut state = state.write().await;
let sealed_generation = if state.memtable.batch_count() == 0 {
None
} else {
let generation = state.memtable.generation();
writer_state.freeze_memtable(&mut state)?;
Some(generation)
};
Ok(SealFence {
sealed_generation,
watchers: state.frozen_flush_watchers.iter().cloned().collect(),
})
}
WriterMode::WalOnly { .. } => Err(Error::invalid_input(
"force_seal_active not available in WAL-only mode (no MemTable)",
)),
}
}
#[instrument(name = "sw_wait_for_flush_drain", level = "info", skip_all, fields(shard_id = %self.config.shard_id, epoch = self.epoch))]
pub async fn wait_for_flush_drain(&self) -> Result<()> {
let state_lock = match &self.mode {
WriterMode::MemTable { state, .. } => state,
WriterMode::WalOnly { .. } => {
return Err(Error::invalid_input(
"wait_for_flush_drain not available in WAL-only mode (no MemTable)",
));
}
};
loop {
let watchers: Vec<DurabilityWatcher> = {
let st = state_lock.read().await;
st.frozen_flush_watchers.iter().cloned().collect()
};
if watchers.is_empty() {
return Ok(());
}
for mut w in watchers {
match w.await_value().await {
Some(durability) => durability.into_result()?,
None => {
return Err(Error::io(
"MemTable flush handler exited before reporting completion",
));
}
}
}
}
}
#[instrument(name = "sw_abort", level = "info", skip_all, fields(shard_id = %self.config.shard_id, epoch = self.epoch))]
pub async fn abort(&self) -> Result<()> {
info!(
"Aborting ShardWriter for shard {} (no flush)",
self.config.shard_id
);
self.task_executor.shutdown_all().await?;
Ok(())
}
async fn flush_final_wal(
wal_flush_tx: &mpsc::UnboundedSender<TriggerWalFlush>,
source: WalFlushSource,
end_batch_position: usize,
) -> Result<()> {
let done = WatchableOnceCell::new();
let mut reader = done.reader();
if wal_flush_tx
.send(TriggerWalFlush {
source,
end_batch_position,
done: Some(done),
})
.is_err()
{
return Err(Error::io("WAL flush channel closed during close"));
}
match reader.await_value().await {
Some(Ok(_)) => Ok(()),
Some(Err(failure)) => Err(failure.into_error()),
None => Err(Error::io(
"WAL flush handler exited before reporting durability during close",
)),
}
}
fn merge_close_stage(
close_result: Result<()>,
stage: &str,
stage_result: Result<()>,
) -> Result<()> {
if let (Err(_), Err(stage_error)) = (&close_result, &stage_result) {
warn!("Close stage '{stage}' also failed: {stage_error}");
}
close_result.and(stage_result)
}
#[instrument(name = "sw_close", level = "info", skip_all, fields(shard_id = %self.config.shard_id, epoch = self.epoch))]
pub async fn close(self) -> Result<()> {
info!("Closing ShardWriter for shard {}", self.config.shard_id);
let mut close_result: Result<()> = Ok(());
match &self.mode {
WriterMode::MemTable {
state,
writer_state,
..
} => {
let st = state.read().await;
let batch_store = st.memtable.batch_store();
let indexes = st.memtable.indexes_arc();
let batch_count = st.memtable.batch_count();
drop(st);
if batch_count > 0
&& let Some(indexes) = indexes
&& indexes.indexed_count() < batch_count
{
let mut watcher = self.wal_flusher.track_batch(
Some(Arc::clone(&indexes)),
batch_count,
0, );
writer_state.trigger_index_apply(
Arc::clone(&batch_store),
indexes,
batch_count,
)?;
watcher.wait().await?;
}
if batch_count > 0 {
let stage_result = Self::flush_final_wal(
&writer_state.wal_flush_tx,
WalFlushSource::BatchStore { batch_store },
batch_count,
)
.await;
close_result =
Self::merge_close_stage(close_result, "final WAL flush", stage_result);
}
let watchers: Vec<_> = {
let mut st = state.write().await;
if st.memtable.row_count() > 0 {
let freeze_result = writer_state.freeze_memtable(&mut st).map(|_| ());
close_result = Self::merge_close_stage(
close_result,
"active MemTable freeze",
freeze_result,
);
}
st.frozen_flush_watchers.iter().cloned().collect()
};
for mut watcher in watchers {
let stage_result = match watcher.await_value().await {
Some(durability) => durability.into_result(),
None => Err(Error::io(
"MemTable flush handler exited before reporting completion during close",
)),
};
close_result = Self::merge_close_stage(
close_result,
"frozen MemTable flush watcher",
stage_result,
);
}
}
WriterMode::WalOnly {
state,
wal_flush_tx,
trigger: _,
backpressure: _,
} => {
let pending = state.batch_count();
let end_position = state.next_batch_position();
if pending > 0 {
let stage_result = Self::flush_final_wal(
wal_flush_tx,
WalFlushSource::WalOnly {
state: state.clone(),
},
end_position,
)
.await;
close_result =
Self::merge_close_stage(close_result, "final WAL flush", stage_result);
}
}
}
let shutdown_result = self.task_executor.shutdown_all().await;
let close_result = Self::merge_close_stage(close_result, "task shutdown", shutdown_result);
match &close_result {
Ok(()) => info!("ShardWriter closed for shard {}", self.config.shard_id),
Err(error) => warn!(
"ShardWriter close for shard {} failed: {error}",
self.config.shard_id
),
}
close_result
}
}
#[derive(Debug, Clone)]
pub struct MemTableStats {
pub row_count: usize,
pub batch_count: usize,
pub generation: u64,
pub max_buffered_batch_position: Option<usize>,
pub durable_batch_count: usize,
pub global_offset: usize,
pub pending_wal_start_batch_position: Option<usize>,
pub pending_wal_end_batch_position: Option<usize>,
pub pending_wal_batch_count: usize,
pub pending_wal_row_count: usize,
pub pending_wal_estimated_bytes: usize,
pub frozen_count: usize,
}
#[derive(Debug, Clone)]
pub struct WalStats {
pub next_wal_entry_position: u64,
}
#[derive(Debug)]
pub struct SealFence {
sealed_generation: Option<u64>,
watchers: Vec<DurabilityWatcher>,
}
impl SealFence {
pub fn sealed_generation(&self) -> Option<u64> {
self.sealed_generation
}
pub async fn wait(self) -> Result<()> {
for mut watcher in self.watchers {
match watcher.await_value().await {
Some(durability) => durability.into_result()?,
None => {
return Err(Error::io(
"MemTable flush handler exited before reporting completion",
));
}
}
}
Ok(())
}
}
fn next_pending_store(
frozen: impl Iterator<Item = Arc<BatchStore>>,
active: Arc<BatchStore>,
durable: usize,
) -> Option<Arc<BatchStore>> {
frozen
.chain(std::iter::once(active))
.find(|store| store.global_end() > durable)
}
struct IndexApplyHandler {
cursors: Arc<WriterCursors>,
wal_flusher: Arc<WalFlusher>,
stats: SharedWriteStats,
}
#[async_trait]
impl MessageHandler<TriggerIndexApply> for IndexApplyHandler {
async fn handle(&mut self, message: TriggerIndexApply) -> Result<()> {
match apply_index_range(&self.cursors, message).await {
Ok(applied) => {
if applied.rows_indexed > 0 {
self.stats
.record_index_update(applied.duration, applied.rows_indexed);
}
Ok(())
}
Err(e) => {
self.wal_flusher.poison(&e);
Err(e)
}
}
}
}
struct WalFlushHandler {
wal_flusher: Arc<WalFlusher>,
memtable_state: Option<Arc<RwLock<WriterState>>>,
wal_only_state: Option<Arc<WalOnlyState>>,
flush_interval: Option<Duration>,
stats: SharedWriteStats,
observer: Option<Arc<dyn WalObserver>>,
}
impl WalFlushHandler {
fn new(
wal_flusher: Arc<WalFlusher>,
memtable_state: Option<Arc<RwLock<WriterState>>>,
wal_only_state: Option<Arc<WalOnlyState>>,
flush_interval: Option<Duration>,
stats: SharedWriteStats,
observer: Option<Arc<dyn WalObserver>>,
) -> Self {
Self {
wal_flusher,
memtable_state,
wal_only_state,
flush_interval,
stats,
observer,
}
}
}
#[async_trait]
impl MessageHandler<TriggerWalFlush> for WalFlushHandler {
fn tickers(&mut self) -> Vec<(Duration, MessageFactory<TriggerWalFlush>)> {
let Some(interval) = self.flush_interval.filter(|d| !d.is_zero()) else {
return vec![];
};
vec![(
interval,
Box::new(|| TriggerWalFlush {
source: WalFlushSource::NextPending,
end_batch_position: 0,
done: None,
}),
)]
}
async fn handle(&mut self, message: TriggerWalFlush) -> Result<()> {
let TriggerWalFlush {
source,
end_batch_position,
done,
} = message;
let (source, end_batch_position) = match source {
WalFlushSource::NextPending => match self.resolve_next_pending().await {
Some(resolved) => resolved,
None => return Ok(()),
},
other => (other, end_batch_position),
};
let result = self.do_flush(source, end_batch_position).await;
if let (Ok(flush_result), Some(state_lock)) = (&result, &self.memtable_state)
&& let Some(entry) = &flush_result.entry
{
let mut state = state_lock.write().await;
state.last_flushed_wal_entry_position =
state.last_flushed_wal_entry_position.max(entry.position);
}
if let Some(cell) = done {
cell.write(result.map_err(|e| WalFlushFailure::from_error(&e)));
}
Ok(())
}
}
impl WalFlushHandler {
async fn resolve_next_pending(&self) -> Option<(WalFlushSource, usize)> {
if let Some(state_lock) = self.memtable_state.as_ref() {
let state = state_lock.read().await;
let durable = self.wal_flusher.durable();
return next_pending_store(
state
.frozen_memtables
.iter()
.map(|frozen| frozen.memtable.batch_store()),
state.memtable.batch_store(),
durable,
)
.map(|store| {
let end = store.len();
(WalFlushSource::BatchStore { batch_store: store }, end)
});
}
let state = self.wal_only_state.as_ref()?;
if state.batch_count() == 0 {
return None;
}
let end = state.next_batch_position();
Some((
WalFlushSource::WalOnly {
state: Arc::clone(state),
},
end,
))
}
#[instrument(
name = "wal_do_flush",
level = "debug",
skip_all,
fields(end_batch_position)
)]
async fn do_flush(
&self,
source: WalFlushSource,
end_batch_position: usize,
) -> Result<WalFlushResult> {
let start = Instant::now();
if let WalFlushSource::BatchStore { batch_store, .. } = &source {
let flushed_up_to = batch_store.local_end(self.wal_flusher.durable());
let is_frozen_flush = if let Some(state_lock) = &self.memtable_state {
let state = state_lock.read().await;
!Arc::ptr_eq(batch_store, &state.memtable.batch_store())
} else {
false
};
if !is_frozen_flush && flushed_up_to >= end_batch_position {
return Ok(empty_flush_result());
}
}
let flush_result = self.wal_flusher.flush(&source, end_batch_position).await?;
let batches_flushed = flush_result
.entry
.as_ref()
.map(|e| e.num_batches)
.unwrap_or(0);
if batches_flushed > 0 {
let elapsed = start.elapsed();
self.stats.record_wal_flush(elapsed, flush_result.wal_bytes);
self.stats.record_wal_io(flush_result.wal_io_duration);
if let Some(observer) = &self.observer {
observer.on_wal_flush(elapsed, flush_result.wal_bytes);
}
}
Ok(flush_result)
}
}
struct MemTableFlushHandler {
state: Arc<RwLock<WriterState>>,
memory: Arc<ArcSwap<ResidentMemTables>>,
flusher: Arc<MemTableFlusher>,
wal_flusher: Arc<WalFlusher>,
epoch: u64,
index_configs: Vec<MemIndexConfig>,
stats: SharedWriteStats,
observer: Option<Arc<dyn WalObserver>>,
grace: Duration,
}
impl MemTableFlushHandler {
#[allow(clippy::too_many_arguments)]
fn new(
state: Arc<RwLock<WriterState>>,
memory: Arc<ArcSwap<ResidentMemTables>>,
flusher: Arc<MemTableFlusher>,
wal_flusher: Arc<WalFlusher>,
epoch: u64,
index_configs: Vec<MemIndexConfig>,
stats: SharedWriteStats,
observer: Option<Arc<dyn WalObserver>>,
grace: Duration,
) -> Self {
Self {
state,
memory,
flusher,
wal_flusher,
epoch,
index_configs,
stats,
observer,
grace,
}
}
}
async fn sweep_expired_frozen(
state: &Arc<RwLock<WriterState>>,
memory: &Arc<ArcSwap<ResidentMemTables>>,
grace: Duration,
) {
let now = now_millis();
let grace_ms = grace.as_millis() as u64;
let mut state = state.write().await;
let before = state.frozen_memtables.len();
state
.frozen_memtables
.retain(|frozen| match frozen.flushed_at_ms {
Some(flushed_at) => now.saturating_sub(flushed_at) < grace_ms,
None => true,
});
if state.frozen_memtables.len() != before {
publish_memory(memory, &state);
}
}
#[async_trait]
impl MessageHandler<TriggerMemTableFlush> for MemTableFlushHandler {
async fn handle(&mut self, message: TriggerMemTableFlush) -> Result<()> {
match message {
TriggerMemTableFlush::Flush { memtable, done } => {
let result = self.flush_memtable(memtable).await;
if let Some(tx) = done {
let _ = tx.send(result);
} else {
result?;
}
}
}
Ok(())
}
}
impl MemTableFlushHandler {
#[instrument(name = "mt_flush", level = "info", skip_all, fields(generation = memtable.generation(), row_count = memtable.row_count()))]
async fn flush_memtable(
&mut self,
memtable: Arc<MemTable>,
) -> Result<super::memtable::flush::FlushResult> {
let start = Instant::now();
let flush_result = async {
let wal_flushed_position =
if let Some(mut completion_reader) = memtable.take_wal_flush_completion() {
match completion_reader.await_value().await {
Some(Ok(flush_result)) => flush_result.entry.map(|e| e.position),
Some(Err(e)) => return Err(e.into_error()),
None => {
return Err(Error::io(
"WAL flush handler exited before reporting completion",
));
}
}
} else {
None
};
if !self.index_configs.is_empty()
&& let Some(indexes) = memtable.indexes_arc()
{
let target_indexed = memtable.batch_count();
self.wal_flusher
.track_batch(Some(indexes), target_indexed, 0)
.wait()
.await?;
}
let covered_wal_entry_position = wal_flushed_position
.or_else(|| memtable.frozen_at_wal_entry_position())
.unwrap_or(0);
let durable = self.wal_flusher.durable();
if self.index_configs.is_empty() {
self.flusher
.flush(&memtable, self.epoch, covered_wal_entry_position, durable)
.await
} else {
Box::pin(self.flusher.flush_with_indexes(
&memtable,
self.epoch,
&self.index_configs,
covered_wal_entry_position,
durable,
))
.await
}
}
.await;
let durability = match &flush_result {
Ok(_) => DurabilityResult::ok(),
Err(e) => DurabilityResult::err(e.to_string()),
};
memtable.signal_memtable_flush_complete(durability);
{
let mut state = self.state.write().await;
state.frozen_flush_watchers.pop_front();
if flush_result.is_ok() {
let sstable = memtable.generation();
if self.grace.is_zero() {
state
.frozen_memtables
.retain(|frozen| frozen.memtable.generation() != sstable);
} else {
let now = now_millis();
for frozen in state.frozen_memtables.iter_mut() {
if frozen.memtable.generation() == sstable {
frozen.flushed_at_ms = Some(now);
}
}
}
}
publish_memory(&self.memory, &state);
}
let result = flush_result?;
let elapsed = start.elapsed();
self.stats
.record_memtable_flush(elapsed, result.rows_flushed);
if let Some(observer) = &self.observer {
observer.on_memtable_flush(elapsed, result.rows_flushed);
}
info!(
"Flushed frozen memtable generation {} ({} rows in {:?})",
result.sstable.generation,
result.rows_flushed,
start.elapsed()
);
Ok(result)
}
}
#[derive(Debug, Default)]
pub struct WriteStats {
put_count: AtomicU64,
put_time_nanos: AtomicU64,
wal_flush_count: AtomicU64,
wal_flush_time_nanos: AtomicU64,
wal_flush_bytes: AtomicU64,
wal_io_time_nanos: AtomicU64,
wal_io_count: AtomicU64,
index_update_time_nanos: AtomicU64,
index_update_count: AtomicU64,
index_update_rows: AtomicU64,
memtable_flush_count: AtomicU64,
memtable_flush_time_nanos: AtomicU64,
memtable_flush_rows: AtomicU64,
}
#[derive(Debug, Clone)]
pub struct WriteStatsSnapshot {
pub put_count: u64,
pub put_time: Duration,
pub wal_flush_count: u64,
pub wal_flush_time: Duration,
pub wal_flush_bytes: u64,
pub wal_io_time: Duration,
pub wal_io_count: u64,
pub index_update_time: Duration,
pub index_update_count: u64,
pub index_update_rows: u64,
pub memtable_flush_count: u64,
pub memtable_flush_time: Duration,
pub memtable_flush_rows: u64,
}
impl WriteStats {
pub fn new() -> Self {
Self::default()
}
pub fn record_put(&self, duration: Duration) {
self.put_count.fetch_add(1, Ordering::Relaxed);
self.put_time_nanos
.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
}
pub fn record_wal_flush(&self, duration: Duration, bytes: usize) {
self.wal_flush_count.fetch_add(1, Ordering::Relaxed);
self.wal_flush_time_nanos
.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
self.wal_flush_bytes
.fetch_add(bytes as u64, Ordering::Relaxed);
}
pub fn record_wal_io(&self, duration: Duration) {
self.wal_io_count.fetch_add(1, Ordering::Relaxed);
self.wal_io_time_nanos
.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
}
pub fn record_index_update(&self, duration: Duration, rows: usize) {
self.index_update_count.fetch_add(1, Ordering::Relaxed);
self.index_update_time_nanos
.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
self.index_update_rows
.fetch_add(rows as u64, Ordering::Relaxed);
}
pub fn record_memtable_flush(&self, duration: Duration, rows: usize) {
self.memtable_flush_count.fetch_add(1, Ordering::Relaxed);
self.memtable_flush_time_nanos
.fetch_add(duration.as_nanos() as u64, Ordering::Relaxed);
self.memtable_flush_rows
.fetch_add(rows as u64, Ordering::Relaxed);
}
pub fn snapshot(&self) -> WriteStatsSnapshot {
WriteStatsSnapshot {
put_count: self.put_count.load(Ordering::Relaxed),
put_time: Duration::from_nanos(self.put_time_nanos.load(Ordering::Relaxed)),
wal_flush_count: self.wal_flush_count.load(Ordering::Relaxed),
wal_flush_time: Duration::from_nanos(self.wal_flush_time_nanos.load(Ordering::Relaxed)),
wal_flush_bytes: self.wal_flush_bytes.load(Ordering::Relaxed),
wal_io_time: Duration::from_nanos(self.wal_io_time_nanos.load(Ordering::Relaxed)),
wal_io_count: self.wal_io_count.load(Ordering::Relaxed),
index_update_time: Duration::from_nanos(
self.index_update_time_nanos.load(Ordering::Relaxed),
),
index_update_count: self.index_update_count.load(Ordering::Relaxed),
index_update_rows: self.index_update_rows.load(Ordering::Relaxed),
memtable_flush_count: self.memtable_flush_count.load(Ordering::Relaxed),
memtable_flush_time: Duration::from_nanos(
self.memtable_flush_time_nanos.load(Ordering::Relaxed),
),
memtable_flush_rows: self.memtable_flush_rows.load(Ordering::Relaxed),
}
}
pub fn reset(&self) {
self.put_count.store(0, Ordering::Relaxed);
self.put_time_nanos.store(0, Ordering::Relaxed);
self.wal_flush_count.store(0, Ordering::Relaxed);
self.wal_flush_time_nanos.store(0, Ordering::Relaxed);
self.wal_flush_bytes.store(0, Ordering::Relaxed);
self.wal_io_time_nanos.store(0, Ordering::Relaxed);
self.wal_io_count.store(0, Ordering::Relaxed);
self.index_update_time_nanos.store(0, Ordering::Relaxed);
self.index_update_count.store(0, Ordering::Relaxed);
self.index_update_rows.store(0, Ordering::Relaxed);
self.memtable_flush_count.store(0, Ordering::Relaxed);
self.memtable_flush_time_nanos.store(0, Ordering::Relaxed);
self.memtable_flush_rows.store(0, Ordering::Relaxed);
}
}
impl WriteStatsSnapshot {
pub fn avg_put_latency(&self) -> Option<Duration> {
if self.put_count > 0 {
Some(self.put_time / self.put_count as u32)
} else {
None
}
}
pub fn put_throughput(&self) -> f64 {
if self.put_time.as_secs_f64() > 0.0 {
self.put_count as f64 / self.put_time.as_secs_f64()
} else {
0.0
}
}
pub fn avg_wal_flush_latency(&self) -> Option<Duration> {
if self.wal_flush_count > 0 {
Some(self.wal_flush_time / self.wal_flush_count as u32)
} else {
None
}
}
pub fn avg_wal_flush_bytes(&self) -> Option<u64> {
self.wal_flush_bytes.checked_div(self.wal_flush_count)
}
pub fn wal_throughput_bytes(&self) -> f64 {
if self.wal_flush_time.as_secs_f64() > 0.0 {
self.wal_flush_bytes as f64 / self.wal_flush_time.as_secs_f64()
} else {
0.0
}
}
pub fn avg_wal_io_latency(&self) -> Option<Duration> {
if self.wal_io_count > 0 {
Some(self.wal_io_time / self.wal_io_count as u32)
} else {
None
}
}
pub fn avg_index_update_latency(&self) -> Option<Duration> {
if self.index_update_count > 0 {
Some(self.index_update_time / self.index_update_count as u32)
} else {
None
}
}
pub fn avg_index_update_rows(&self) -> Option<u64> {
self.index_update_rows.checked_div(self.index_update_count)
}
pub fn avg_memtable_flush_latency(&self) -> Option<Duration> {
if self.memtable_flush_count > 0 {
Some(self.memtable_flush_time / self.memtable_flush_count as u32)
} else {
None
}
}
pub fn avg_memtable_flush_rows(&self) -> Option<u64> {
self.memtable_flush_rows
.checked_div(self.memtable_flush_count)
}
pub fn log_summary(&self, prefix: &str) {
tracing::info!(
prefix = prefix,
put_count = self.put_count,
put_throughput = self.put_throughput(),
put_avg_latency_us = self.avg_put_latency().unwrap_or_default().as_micros() as u64,
wal_flush_count = self.wal_flush_count,
wal_flush_bytes = self.wal_flush_bytes,
wal_avg_latency_us =
self.avg_wal_flush_latency().unwrap_or_default().as_micros() as u64,
memtable_flush_count = self.memtable_flush_count,
memtable_flush_rows = self.memtable_flush_rows,
memtable_avg_latency_us = self
.avg_memtable_flush_latency()
.unwrap_or_default()
.as_micros() as u64,
"MemWAL stats summary"
);
}
pub fn log_wal_breakdown(&self, prefix: &str) {
if self.wal_flush_count > 0 {
tracing::info!(
prefix = prefix,
wal_total_latency_us =
self.avg_wal_flush_latency().unwrap_or_default().as_micros() as u64,
wal_io_latency_us =
self.avg_wal_io_latency().unwrap_or_default().as_micros() as u64,
index_update_latency_us = self
.avg_index_update_latency()
.unwrap_or_default()
.as_micros() as u64,
index_update_rows = self.index_update_rows,
"MemWAL WAL flush breakdown"
);
}
}
}
pub type SharedWriteStats = Arc<WriteStats>;
pub fn new_shared_stats() -> SharedWriteStats {
Arc::new(WriteStats::new())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::dataset::mem_wal::test_util::failing_memory_store;
use arrow_array::{FixedSizeListArray, Float32Array, Int32Array, StringArray};
use arrow_schema::{DataType, Field};
use lance_core::FenceReason;
use rstest::rstest;
use std::sync::atomic::AtomicUsize;
use tempfile::TempDir;
async fn create_local_store() -> (Arc<ObjectStore>, Path, String, TempDir) {
let temp_dir = tempfile::tempdir().unwrap();
let uri = format!("file://{}", temp_dir.path().display());
let (store, path) = ObjectStore::from_uri(&uri).await.unwrap();
(store, path, uri, temp_dir)
}
#[test]
fn test_merge_close_stage_preserves_first_error() {
let result = ShardWriter::merge_close_stage(
Err(Error::io("primary close error")),
"secondary close stage",
Err(Error::io("secondary close error")),
);
let error = result.expect_err("close must preserve the first error");
assert!(matches!(&error, Error::IO { .. }));
assert!(
error.to_string().contains("primary close error"),
"unexpected error: {error}"
);
assert!(
!error.to_string().contains("secondary close error"),
"secondary error replaced the primary error: {error}"
);
}
fn create_pk_test_schema() -> Arc<ArrowSchema> {
let mut id_metadata = std::collections::HashMap::new();
id_metadata.insert(
"lance-schema:unenforced-primary-key".to_string(),
"true".to_string(),
);
let id = Field::new("id", DataType::Int32, false).with_metadata(id_metadata);
Arc::new(ArrowSchema::new(vec![
id,
Field::new("name", DataType::Utf8, true),
]))
}
fn create_strict_pk_test_schema() -> Arc<ArrowSchema> {
let fields: Vec<Field> = create_pk_test_schema()
.fields()
.iter()
.map(|f| f.as_ref().clone().with_nullable(false))
.collect();
Arc::new(ArrowSchema::new(fields))
}
fn id_only_keys(ids: &[i32]) -> RecordBatch {
RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![Field::new(
"id",
DataType::Int32,
false,
)])),
vec![Arc::new(Int32Array::from(ids.to_vec()))],
)
.unwrap()
}
#[tokio::test]
async fn test_put_rejects_swapped_same_typed_columns() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = Arc::new(ArrowSchema::new(vec![
Field::new("name", DataType::Utf8, true),
Field::new("email", DataType::Utf8, true),
]));
let writer = ShardWriter::open(
store,
base_path,
base_uri,
ShardWriterConfig {
shard_id: Uuid::new_v4(),
..Default::default()
},
schema,
vec![],
)
.await
.unwrap();
let swapped = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![
Field::new("email", DataType::Utf8, true),
Field::new("name", DataType::Utf8, true),
])),
vec![
Arc::new(StringArray::from(vec!["a@example.com"])),
Arc::new(StringArray::from(vec!["a"])),
],
)
.unwrap();
let error = writer.put(vec![swapped]).await.unwrap_err();
assert!(
matches!(error, Error::InvalidInput { .. }),
"expected InvalidInput, got {error:?}"
);
let message = error.to_string();
assert!(
message.contains("column 0") && message.contains("email") && message.contains("name"),
"error should name the position and both columns: {message}"
);
writer.close().await.unwrap();
}
#[test]
fn test_ensure_tombstone_column_injects_false() {
let base = create_test_schema();
let storage = schema_with_tombstone(&base);
let out = ensure_tombstone_column(create_test_batch(&base, 0, 3), &storage).unwrap();
assert_eq!(out.schema(), storage);
let ts = out
.column_by_name(TOMBSTONE)
.unwrap()
.as_any()
.downcast_ref::<BooleanArray>()
.unwrap();
assert!(
(0..3).all(|i| !ts.value(i)),
"put injects _tombstone = false"
);
let again = ensure_tombstone_column(out.clone(), &storage).unwrap();
assert_eq!(again.schema(), out.schema());
}
#[test]
fn test_build_tombstone_batch_shape() {
let storage = schema_with_tombstone(&create_test_schema());
let tomb =
build_tombstone_batch(&id_only_keys(&[5, 7]), &storage, &["id".to_string()]).unwrap();
assert_eq!(tomb.schema(), storage);
assert_eq!(tomb.num_rows(), 2);
let ids = tomb
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
assert_eq!(ids.values(), &[5, 7]);
let name = tomb.column_by_name("name").unwrap();
assert!(
name.is_null(0) && name.is_null(1),
"non-PK columns are null in a tombstone"
);
let ts = tomb
.column_by_name(TOMBSTONE)
.unwrap()
.as_any()
.downcast_ref::<BooleanArray>()
.unwrap();
assert!(ts.value(0) && ts.value(1), "_tombstone = true");
}
#[test]
fn test_build_tombstone_batch_missing_pk_errors() {
let storage = schema_with_tombstone(&create_test_schema());
let keys = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![Field::new(
"other",
DataType::Int32,
false,
)])),
vec![Arc::new(Int32Array::from(vec![1]))],
)
.unwrap();
assert!(build_tombstone_batch(&keys, &storage, &["id".to_string()]).is_err());
}
#[test]
fn test_build_tombstone_batch_nulls_non_nullable_base_column() {
let pk = ["id".to_string()];
let base = Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("v", DataType::Int32, false),
]));
let storage = relax_non_pk_nullability(&schema_with_tombstone(&base), &pk);
let batch = build_tombstone_batch(&id_only_keys(&[1]), &storage, &pk).unwrap();
assert!(batch["v"].is_null(0), "the tombstone must null `v`");
assert!(!batch["id"].is_null(0), "the primary key survives");
assert!(
batch[TOMBSTONE]
.as_any()
.downcast_ref::<BooleanArray>()
.unwrap()
.value(0)
);
}
#[test]
fn test_build_tombstone_batch_rejects_null_primary_key() {
let pk = ["id".to_string()];
let base = Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("v", DataType::Int32, false),
]));
let storage = relax_non_pk_nullability(&schema_with_tombstone(&base), &pk);
let keys = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![Field::new(
"id",
DataType::Int32,
true,
)])),
vec![Arc::new(Int32Array::from(vec![None::<i32>]))],
)
.unwrap();
let error = build_tombstone_batch(&keys, &storage, &pk).unwrap_err();
assert!(
matches!(error, Error::InvalidInput { .. }),
"expected InvalidInput, got {error:?}"
);
assert!(
error.to_string().contains("non-nullable"),
"error should name the nullability violation: {error}"
);
}
#[tokio::test]
async fn test_shard_writer_delete_round_trip() {
use crate::dataset::mem_wal::scanner::LsmScanner;
use futures::TryStreamExt;
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: true,
..Default::default()
};
let shard_id = config.shard_id;
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
config,
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer.delete(vec![id_only_keys(&[2])]).await.unwrap();
let refs = writer.in_memory_memtable_refs().await.unwrap();
let scanner = LsmScanner::without_base_table(
schema.clone(),
base_uri,
vec![],
vec!["id".to_string()],
)
.with_in_memory_memtables(shard_id, refs);
let batches: Vec<RecordBatch> = scanner
.try_into_stream()
.await
.unwrap()
.try_collect()
.await
.unwrap();
let mut ids: Vec<i32> = Vec::new();
for b in &batches {
let arr = b
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
ids.extend((0..arr.len()).map(|i| arr.value(i)));
}
ids.sort_unstable();
assert_eq!(
ids,
vec![0, 1, 3, 4],
"id=2 deleted; tombstone not surfaced"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_delete_against_non_nullable_base_column_round_trip() {
use crate::dataset::mem_wal::scanner::LsmScanner;
use futures::TryStreamExt;
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_strict_pk_test_schema();
assert!(
!schema.field_with_name("name").unwrap().is_nullable(),
"the point of this test is a non-nullable non-PK column"
);
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: true,
..Default::default()
};
let shard_id = config.shard_id;
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
config,
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer.delete(vec![id_only_keys(&[2])]).await.unwrap();
let refs = writer.in_memory_memtable_refs().await.unwrap();
let scanner = LsmScanner::without_base_table(
schema.clone(),
base_uri,
vec![],
vec!["id".to_string()],
)
.with_in_memory_memtables(shard_id, refs);
let batches: Vec<RecordBatch> = scanner
.try_into_stream()
.await
.unwrap()
.try_collect()
.await
.unwrap();
let mut rows: Vec<(i32, String)> = Vec::new();
for b in &batches {
assert!(
!b.schema().field_with_name("name").unwrap().is_nullable(),
"egress must narrow back to the logical schema"
);
let ids = b["id"].as_any().downcast_ref::<Int32Array>().unwrap();
let names = b["name"].as_any().downcast_ref::<StringArray>().unwrap();
rows.extend((0..ids.len()).map(|i| (ids.value(i), names.value(i).to_string())));
}
rows.sort_unstable();
assert_eq!(
rows,
vec![
(0, "name_0".to_string()),
(1, "name_1".to_string()),
(3, "name_3".to_string()),
(4, "name_4".to_string()),
],
"id=2 deleted; every survivor keeps its non-nullable value"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_put_rejects_null_in_non_nullable_base_column() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_strict_pk_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
ShardWriterConfig {
shard_id: Uuid::new_v4(),
..Default::default()
},
schema.clone(),
vec![],
)
.await
.unwrap();
let error = writer.put(vec![null_name_batch()]).await.unwrap_err();
assert!(
matches!(error, Error::InvalidInput { .. }),
"expected InvalidInput, got {error:?}"
);
assert!(
error.to_string().contains("base table schema"),
"error should point at the schema contract: {error}"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_put_rejects_null_in_non_nullable_base_column() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_strict_pk_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(Uuid::new_v4()),
schema.clone(),
vec![],
)
.await
.unwrap();
let error = writer.put(vec![null_name_batch()]).await.unwrap_err();
assert!(
matches!(error, Error::InvalidInput { .. }),
"expected InvalidInput, got {error:?}"
);
writer.close().await.unwrap();
}
fn null_name_batch() -> RecordBatch {
RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("name", DataType::Utf8, true),
])),
vec![
Arc::new(Int32Array::from(vec![0, 1])),
Arc::new(StringArray::from(vec![Some("a"), None])),
],
)
.unwrap()
}
#[tokio::test]
async fn test_shard_writer_delete_no_wait_durable_visible_after_watcher() {
use crate::dataset::mem_wal::scanner::LsmScanner;
use futures::TryStreamExt;
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: true,
..Default::default()
};
let shard_id = config.shard_id;
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
config,
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
let (_result, watcher) = writer
.delete_no_wait(vec![id_only_keys(&[2])])
.await
.unwrap();
assert_eq!(writer.memtable_stats().await.unwrap().row_count, 6);
let mut watcher = watcher.expect("durable_write returns a watcher");
watcher.wait().await.unwrap();
let refs = writer.in_memory_memtable_refs().await.unwrap();
let scanner = LsmScanner::without_base_table(
schema.clone(),
base_uri,
vec![],
vec!["id".to_string()],
)
.with_in_memory_memtables(shard_id, refs);
let batches: Vec<RecordBatch> = scanner
.try_into_stream()
.await
.unwrap()
.try_collect()
.await
.unwrap();
let mut ids: Vec<i32> = Vec::new();
for b in &batches {
let arr = b
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
ids.extend((0..arr.len()).map(|i| arr.value(i)));
}
ids.sort_unstable();
assert_eq!(
ids,
vec![0, 1, 3, 4],
"delete folded by the LSM read once the watcher resolves"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_non_durable_delete_is_read_your_writes() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
let (_result, watcher) = writer
.delete_no_wait(vec![id_only_keys(&[2])])
.await
.unwrap();
let mut watcher = watcher.expect("a non-durable delete awaits its index apply");
watcher.wait().await.unwrap();
assert_eq!(writer.memtable_stats().await.unwrap().row_count, 6);
writer.close().await.unwrap();
}
async fn read_sstable_ids_via_lsm(
writer: &ShardWriter,
schema: Arc<ArrowSchema>,
base_uri: &str,
shard_id: Uuid,
filter: Option<&str>,
) -> Vec<i32> {
use crate::dataset::mem_wal::scanner::{LsmScanner, ShardSnapshot};
use futures::TryStreamExt;
let manifest = writer.manifest().await.unwrap().unwrap();
let mut snapshot =
ShardSnapshot::new(shard_id).with_current_generation(manifest.current_generation);
for sstable in &manifest.sstables {
snapshot = snapshot.with_sstable(sstable.generation, sstable.path.clone());
}
let mut scanner = LsmScanner::without_base_table(
schema,
base_uri.to_string(),
vec![snapshot],
vec!["id".to_string()],
);
if let Some(predicate) = filter {
scanner = scanner.filter(predicate).unwrap();
}
let batches: Vec<RecordBatch> = scanner
.try_into_stream()
.await
.unwrap()
.try_collect()
.await
.unwrap();
let mut ids: Vec<i32> = Vec::new();
for b in &batches {
let arr = b
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
ids.extend((0..arr.len()).map(|i| arr.value(i)));
}
ids.sort_unstable();
ids
}
fn flush_test_config(shard_id: Uuid) -> ShardWriterConfig {
ShardWriterConfig {
shard_id,
durable_write: false,
manifest_scan_batch_size: 2,
..Default::default()
}
}
#[tokio::test]
async fn test_shard_writer_delete_then_flush_round_trip() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let shard_id = Uuid::new_v4();
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
flush_test_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer.delete(vec![id_only_keys(&[2])]).await.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, None).await,
vec![0, 1, 3, 4],
"id=2 deleted before flush; tombstone must not surface in an SSTable scan"
);
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, Some("id < 3"))
.await,
vec![0, 1],
"filtered read after flush must not resurface deleted id=2"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_delete_across_sstables() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let shard_id = Uuid::new_v4();
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
flush_test_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
writer.delete(vec![id_only_keys(&[0])]).await.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, None).await,
vec![1, 2, 3, 4],
"id=0 tombstoned in a newer gen must mask the older gen's live row"
);
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, Some("id < 1"))
.await,
Vec::<i32>::new(),
"filtered read 'id < 1' must not resurface cross-gen deleted id=0"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_delete_across_sstables_indexed() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = create_pk_test_schema();
let shard_id = Uuid::new_v4();
let index_configs = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "id_idx".to_string(),
field_id: 0,
column: "id".to_string(),
})];
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
flush_test_config(shard_id),
schema.clone(),
index_configs,
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
writer.delete(vec![id_only_keys(&[0])]).await.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, None).await,
vec![1, 2, 3, 4],
"indexed cross-gen: full scan must mask deleted id=0"
);
assert_eq!(
read_sstable_ids_via_lsm(&writer, schema.clone(), &base_uri, shard_id, Some("id < 1"))
.await,
Vec::<i32>::new(),
"indexed filtered read 'id < 1' must not resurface deleted id=0 (wallop repro)"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_delete_validation_errors() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let no_pk = create_test_schema();
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
..Default::default()
},
no_pk,
vec![],
)
.await
.unwrap();
assert!(
writer.delete(vec![id_only_keys(&[1])]).await.is_err(),
"delete without a primary key must error"
);
assert!(writer.delete(vec![]).await.is_err());
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_indexed_composite_key_delete_removes_row() {
use crate::Dataset;
use crate::dataset::{WhenMatched, WhenNotMatched, WhenNotMatchedBySource};
use crate::index::DatasetIndexExt;
use arrow_array::{Int64Array, RecordBatchIterator};
use lance_index::IndexType;
use lance_index::scalar::ScalarIndexParams;
let schema = Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("shard_key", DataType::Int64, false),
Field::new("value", DataType::Int64, true),
]));
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int64Array::from_iter_values(0..5)),
Arc::new(Int64Array::from_iter_values((0..5).map(|i| i % 8))),
Arc::new(Int64Array::from_iter_values((0..5).map(|i| i * 100))),
],
)
.unwrap();
let uri = format!("shared-memory://phase1-delete-{}/", Uuid::new_v4().simple());
let mut dataset = Dataset::write(
RecordBatchIterator::new([Ok(batch)], schema.clone()),
&uri,
None,
)
.await
.unwrap();
dataset
.create_index(
&["id"],
IndexType::BTree,
Some("id_btree".to_string()),
&ScalarIndexParams::default(),
false,
)
.await
.unwrap();
let keys = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("shard_key", DataType::Int64, false),
])),
vec![
Arc::new(Int64Array::from(vec![0_i64])),
Arc::new(Int64Array::from(vec![0_i64])),
],
)
.unwrap();
let mut builder = crate::dataset::MergeInsertBuilder::try_new(
Arc::new(dataset),
vec!["id".to_string(), "shard_key".to_string()],
)
.unwrap();
builder
.when_matched(WhenMatched::Delete)
.when_not_matched(WhenNotMatched::DoNothing)
.when_not_matched_by_source(WhenNotMatchedBySource::Keep)
.use_index(true);
let job = builder.try_build().unwrap();
let keys_schema = keys.schema();
let (dataset, _stats) = job
.execute_reader(RecordBatchIterator::new([Ok(keys)], keys_schema))
.await
.unwrap();
let all = dataset.scan().try_into_batch().await.unwrap();
let mut ids: Vec<i64> = all
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.values()
.to_vec();
ids.sort_unstable();
assert_eq!(
ids,
vec![1, 2, 3, 4],
"Phase-1 must hard-delete (0, 0) from base"
);
let filtered = dataset
.scan()
.filter("id < 1")
.unwrap()
.try_into_batch()
.await
.unwrap();
assert_eq!(
filtered.num_rows(),
0,
"indexed filter must not resurface the deleted composite key"
);
}
fn create_test_schema() -> Arc<ArrowSchema> {
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("name", DataType::Utf8, true),
]))
}
fn create_test_batch(schema: &ArrowSchema, start_id: i32, num_rows: usize) -> RecordBatch {
RecordBatch::try_new(
Arc::new(schema.clone()),
vec![
Arc::new(Int32Array::from_iter_values(
start_id..start_id + num_rows as i32,
)),
Arc::new(StringArray::from_iter_values(
(0..num_rows).map(|i| format!("name_{}", start_id as usize + i)),
)),
],
)
.unwrap()
}
#[tokio::test]
async fn test_shard_writer_basic_write() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 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 writer = ShardWriter::open(
store,
base_path,
base_uri,
config.clone(),
schema.clone(),
vec![],
)
.await
.unwrap();
let batch = create_test_batch(&schema, 0, 10);
let result = writer.put(vec![batch]).await.unwrap();
assert_eq!(result.batch_positions, 0..1);
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 10);
assert_eq!(stats.batch_count, 1);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_put_no_wait_durable_visible_then_durable() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 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 writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let batch = create_test_batch(&schema, 0, 10);
let (result, watcher) = writer.put_no_wait(vec![batch]).await.unwrap();
assert_eq!(result.batch_positions, 0..1);
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 10);
let mut watcher = watcher.expect("durable_write returns a watcher");
watcher.wait().await.unwrap();
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_non_durable_put_is_read_your_writes() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 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 writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let batch = create_test_batch(&schema, 0, 10);
let (result, watcher) = writer.put_no_wait(vec![batch]).await.unwrap();
assert_eq!(result.batch_positions, 0..1);
let mut watcher = watcher.expect("a non-durable put awaits its index apply");
watcher.wait().await.unwrap();
let scanned = writer.scan().await.unwrap().try_into_batch().await.unwrap();
assert_eq!(
scanned.num_rows(),
10,
"a non-durable put must be readable as soon as it returns"
);
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 10);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_multiple_writes() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 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 writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let batches: Vec<_> = (0..5)
.map(|i| create_test_batch(&schema, i * 10, 10))
.collect();
let result = writer.put(batches).await.unwrap();
assert_eq!(result.batch_positions, 0..5);
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 50);
assert_eq!(stats.batch_count, 5);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_with_indexes() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 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 index_configs = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "id_idx".to_string(),
field_id: 0,
column: "id".to_string(),
})];
let writer = ShardWriter::open(
store,
base_path,
base_uri,
config,
schema.clone(),
index_configs,
)
.await
.unwrap();
let batch = create_test_batch(&schema, 0, 10);
writer.put(vec![batch]).await.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 10);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_sstable_is_indexed() {
use crate::index::DatasetIndexExt;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
shard_id,
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 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 index_configs = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "id_idx".to_string(),
field_id: 0,
column: "id".to_string(),
})];
let writer = ShardWriter::open(
store,
base_path,
base_uri.clone(),
config,
schema.clone(),
index_configs,
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
let manifest = writer.manifest().await.unwrap().unwrap();
assert_eq!(manifest.sstables.len(), 1, "expected exactly one SSTable");
let gen_uri = format!(
"{}/_mem_wal/{}/{}",
base_uri, shard_id, manifest.sstables[0].path
);
let dataset = crate::Dataset::open(&gen_uri).await.unwrap();
let indices = dataset.load_indices().await.unwrap();
assert_eq!(indices.len(), 1, "SSTable should have one index");
assert_eq!(indices[0].name, "id_idx");
let mut scan = dataset.scan();
scan.filter("id = 5").unwrap();
scan.prefilter(true);
let plan = scan.create_plan().await.unwrap();
crate::utils::test::assert_plan_node_equals(
plan,
"LanceRead: ...full_filter=id = Int32(5)...
ScalarIndexQuery: query=[id = 5]@id_idx(BTree)",
)
.await
.unwrap();
let batch = dataset
.scan()
.filter("id = 5")
.unwrap()
.try_into_batch()
.await
.unwrap();
assert_eq!(batch.num_rows(), 1);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_auto_flush_by_size() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 1024, manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
for i in 0..20 {
let batch = create_test_batch(&schema, i * 10, 10);
writer.put(vec![batch]).await.unwrap();
}
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.generation > initial_gen,
"Generation should increment after auto-flush"
);
writer.close().await.unwrap();
}
#[tokio::test(start_paused = true)]
async fn test_saturated_handler_starves_its_ticker_but_not_a_periodic_task() {
#[derive(Debug)]
enum Msg {
Work,
Tick,
}
#[derive(Debug)]
struct SlowHandler {
worked: Arc<AtomicUsize>,
ticked: Arc<AtomicUsize>,
work: Duration,
tick: Duration,
}
#[async_trait]
impl MessageHandler<Msg> for SlowHandler {
fn tickers(&mut self) -> Vec<(Duration, MessageFactory<Msg>)> {
vec![(self.tick, Box::new(|| Msg::Tick))]
}
async fn handle(&mut self, message: Msg) -> Result<()> {
match message {
Msg::Work => {
self.worked.fetch_add(1, Ordering::Relaxed);
tokio::time::sleep(self.work).await;
}
Msg::Tick => {
self.ticked.fetch_add(1, Ordering::Relaxed);
}
}
Ok(())
}
}
let work = Duration::from_millis(100);
let tick = Duration::from_millis(10);
let run = Duration::from_millis(1000);
let expected_ticks = (run.as_millis() / tick.as_millis()) as usize;
let executor = TaskExecutor::new();
let worked = Arc::new(AtomicUsize::new(0));
let ticked = Arc::new(AtomicUsize::new(0));
let swept = Arc::new(AtomicUsize::new(0));
let (tx, rx) = mpsc::unbounded_channel();
executor
.add_handler(
"slow".to_string(),
Box::new(SlowHandler {
worked: worked.clone(),
ticked: ticked.clone(),
work,
tick,
}),
rx,
)
.unwrap();
let swept_by_task = swept.clone();
executor
.add_periodic("sweeper".to_string(), tick, move || {
let swept = swept_by_task.clone();
async move {
swept.fetch_add(1, Ordering::Relaxed);
}
})
.unwrap();
for _ in 0..(run.as_millis() / work.as_millis()) + 2 {
tx.send(Msg::Work).unwrap();
}
tokio::time::sleep(run).await;
let (worked, ticked, swept) = (
worked.load(Ordering::Relaxed),
ticked.load(Ordering::Relaxed),
swept.load(Ordering::Relaxed),
);
assert!(worked > 0, "the handler must have been busy");
assert!(
ticked * 10 < expected_ticks,
"a ticker sharing a saturated handler's loop should be starved, but \
it ran {ticked} of ~{expected_ticks}"
);
assert!(
swept * 2 >= expected_ticks,
"periodic task ran {swept} of ~{expected_ticks}"
);
executor.shutdown_all().await.ok();
}
#[tokio::test]
async fn test_memtable_stats_frozen_count_outlives_frozen_bytes() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 1024, frozen_memtable_grace: Duration::from_secs(600),
manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let fresh = writer.memtable_stats().await.unwrap();
assert_eq!(fresh.frozen_count, 0);
assert_eq!(writer.memory().frozen_bytes(), 0);
for i in 0..20 {
let batch = create_test_batch(&schema, i * 10, 10);
writer.put(vec![batch]).await.unwrap();
}
writer.wait_for_flush_drain().await.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.frozen_count > 0,
"flushed memtables must stay in the read view for the grace window"
);
let memory = writer.memory();
assert_eq!(
memory.frozen_bytes(),
0,
"every seal flushed, so nothing is owed to flush"
);
assert!(
memory.grace_bytes() > 0,
"flushed-but-retained generations hold real memory"
);
assert_eq!(
memory.retained_bytes(),
memory.unflushed_bytes() + memory.grace_bytes(),
"retained is the whole footprint; unflushed is only what a flush can reclaim"
);
writer.close().await.unwrap();
}
#[rstest::rstest]
#[case::memtable(true)]
#[case::wal_only(false)]
#[tokio::test]
async fn test_backpressure_stats_reachable_in_both_modes(#[case] enable_memtable: bool) {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 64 * 1024 * 1024,
manifest_scan_batch_size: 2,
enable_memtable,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let stats = writer.backpressure_stats();
assert_eq!(stats.total_count, 0);
assert_eq!(stats.total_wait_ms, 0);
assert_eq!(stats.active_count, 0);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_task_dispatcher_survives_handle_error() {
use std::sync::atomic::{AtomicUsize, Ordering};
struct FlakyHandler {
call_count: Arc<AtomicUsize>,
}
#[async_trait]
impl MessageHandler<u32> for FlakyHandler {
async fn handle(&mut self, message: u32) -> Result<()> {
let n = self.call_count.fetch_add(1, Ordering::SeqCst);
if n == 0 {
Err(Error::io("first message intentionally fails"))
} else {
let _ = message;
Ok(())
}
}
}
let executor = TaskExecutor::new();
let call_count = Arc::new(AtomicUsize::new(0));
let (tx, rx) = mpsc::unbounded_channel::<u32>();
executor
.add_handler(
"flaky".to_string(),
Box::new(FlakyHandler {
call_count: call_count.clone(),
}),
rx,
)
.unwrap();
tx.send(1).unwrap();
tx.send(2).unwrap();
tx.send(3).unwrap();
for _ in 0..50 {
if call_count.load(Ordering::SeqCst) >= 3 {
break;
}
tokio::time::sleep(tokio::time::Duration::from_millis(20)).await;
}
assert!(
call_count.load(Ordering::SeqCst) >= 3,
"dispatcher exited after the first error; only {} message(s) were handled",
call_count.load(Ordering::SeqCst)
);
executor
.shutdown_all()
.await
.expect("dispatcher should shut down successfully");
}
#[tokio::test]
async fn test_task_executor_shutdown_propagates_cleanup_error_and_joins_all_tasks() {
use std::sync::atomic::{AtomicUsize, Ordering};
struct CleanupHandler {
cleanup_count: Arc<AtomicUsize>,
error_message: Option<&'static str>,
}
#[async_trait]
impl MessageHandler<u32> for CleanupHandler {
async fn handle(&mut self, _message: u32) -> Result<()> {
Ok(())
}
async fn cleanup(&mut self, _shutdown_ok: bool) -> Result<()> {
self.cleanup_count.fetch_add(1, Ordering::SeqCst);
match self.error_message {
Some(message) => Err(Error::io(message)),
None => Ok(()),
}
}
}
let executor = TaskExecutor::new();
let cleanup_count = Arc::new(AtomicUsize::new(0));
let (_failing_tx, failing_rx) = mpsc::unbounded_channel::<u32>();
executor
.add_handler(
"failing-cleanup".to_string(),
Box::new(CleanupHandler {
cleanup_count: cleanup_count.clone(),
error_message: Some("intentional cleanup failure"),
}),
failing_rx,
)
.unwrap();
let (_successful_tx, successful_rx) = mpsc::unbounded_channel::<u32>();
executor
.add_handler(
"successful-cleanup".to_string(),
Box::new(CleanupHandler {
cleanup_count: cleanup_count.clone(),
error_message: None,
}),
successful_rx,
)
.unwrap();
let error = executor
.shutdown_all()
.await
.expect_err("shutdown must propagate the handler cleanup failure");
assert!(matches!(&error, Error::IO { .. }));
assert!(
error.to_string().contains("intentional cleanup failure"),
"unexpected error: {error}"
);
assert_eq!(
cleanup_count.load(Ordering::SeqCst),
2,
"shutdown must join and clean up every task after the first failure"
);
assert!(executor.tasks.read().unwrap().is_empty());
}
#[tokio::test]
async fn test_task_executor_shutdown_propagates_task_panic() {
struct PanickingCleanupHandler;
#[async_trait]
impl MessageHandler<u32> for PanickingCleanupHandler {
async fn handle(&mut self, _message: u32) -> Result<()> {
Ok(())
}
async fn cleanup(&mut self, _shutdown_ok: bool) -> Result<()> {
panic!("intentional cleanup panic");
}
}
let executor = TaskExecutor::new();
let (_tx, rx) = mpsc::unbounded_channel::<u32>();
executor
.add_handler(
"panicking-cleanup".to_string(),
Box::new(PanickingCleanupHandler),
rx,
)
.unwrap();
let error = executor
.shutdown_all()
.await
.expect_err("shutdown must propagate the task panic");
assert!(matches!(&error, Error::Internal { .. }));
assert!(
error.to_string().contains("panicking-cleanup")
&& error.to_string().contains("panicked during shutdown"),
"unexpected error: {error}"
);
}
#[tokio::test]
async fn test_shard_writer_auto_flush_repeatedly_memory_store() {
let base_uri = "memory:///bench_test_flush";
let (store, base_path) = ObjectStore::from_uri(base_uri).await.unwrap();
let base_uri = base_uri.to_string();
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 64,
manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
for i in 0..8 {
let batch = create_test_batch(&schema, i * 10, 10);
writer.put(vec![batch]).await.unwrap();
}
writer.wait_for_flush_drain().await.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.generation >= initial_gen + 3,
"expected repeated successful flushes; generation went {} → {}",
initial_gen,
stats.generation
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_shard_writer_auto_flush_repeatedly_local_store() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 64,
manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
for i in 0..8 {
let batch = create_test_batch(&schema, i * 10, 10);
writer.put(vec![batch]).await.unwrap();
}
writer.wait_for_flush_drain().await.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.generation >= initial_gen + 3,
"expected repeated successful flushes; generation went {} → {}",
initial_gen,
stats.generation
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_close_flushes_active_memtable() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
shard_id,
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: usize::MAX,
max_unflushed_memtable_bytes: usize::MAX,
manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
config.clone(),
schema.clone(),
vec![],
)
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
for i in 0..50 {
let batch = create_test_batch(&schema, i * 10, 10);
writer.put(vec![batch]).await.unwrap();
}
let stats_before = writer.memtable_stats().await.unwrap();
assert_eq!(
stats_before.generation, initial_gen,
"no flush should have fired during puts (size threshold is usize::MAX)"
);
assert!(
stats_before.row_count > 0,
"memtable should hold the rows we just inserted"
);
writer
.close()
.await
.expect("close() must succeed and propagate any freeze/flush error");
let reopened =
ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let manifest = reopened
.manifest()
.await
.unwrap()
.expect("reopened shard must have a persisted manifest");
assert!(
manifest.current_generation > initial_gen,
"expected manifest current_generation to advance past {} after close() flushed the active memtable; got {}",
initial_gen,
manifest.current_generation,
);
reopened.close().await.unwrap();
}
#[rstest]
#[case::memtable(true)]
#[case::wal_only(false)]
#[tokio::test]
async fn test_close_propagates_final_wal_persistence_failure(#[case] enable_memtable: bool) {
let (store, base_path, controls) = failing_memory_store().await;
let base_uri = "memory:///";
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
enable_memtable,
max_wal_buffer_size: usize::MAX,
max_wal_flush_interval: None,
max_wal_persist_retries: 0,
max_memtable_size: usize::MAX,
max_unflushed_memtable_bytes: usize::MAX,
manifest_scan_batch_size: 2,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let task_executor = writer.task_executor.clone();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
controls.fail_wal_puts(usize::MAX);
let error = writer
.close()
.await
.expect_err("close must propagate the final WAL persistence failure");
assert_eq!(error.fence_reason(), Some(FenceReason::PersistenceFailure));
assert!(
error
.to_string()
.contains("injected transient WAL put failure"),
"unexpected error: {error}"
);
assert!(
task_executor.tasks.read().unwrap().is_empty(),
"close must join background tasks before returning an error"
);
}
async fn ground_truth(writer: &ShardWriter) -> (usize, usize) {
let state = writer.memtable_state_lock().unwrap().read().await;
let active = in_memory_ref(&state.memtable).resident_bytes();
let frozen = state
.frozen_memtables
.iter()
.filter(|frozen| frozen.flushed_at_ms.is_none())
.map(|frozen| in_memory_ref(&frozen.memtable).resident_bytes())
.sum();
(active, frozen)
}
async fn assert_no_drift(writer: &ShardWriter, after: &str) {
let (active, frozen) = ground_truth(writer).await;
let memory = writer.memory();
assert_eq!(
(memory.active_bytes(), memory.frozen_bytes()),
(active, frozen),
"published memory drifted from the writer state after {after}"
);
assert_eq!(
memory.unflushed_bytes(),
active + frozen,
"unflushed must be the sum of the two terms after {after}"
);
}
#[rstest]
#[case::grace_keeps_handles(Duration::from_secs(600))]
#[case::zero_grace_evicts_on_commit(Duration::ZERO)]
#[tokio::test]
async fn test_memory_snapshot_never_drifts_from_writer_state(#[case] grace: Duration) {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 2048,
frozen_memtable_grace: grace,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
assert_no_drift(&writer, "open").await;
for i in 0..24 {
writer
.put(vec![create_test_batch(&schema, i * 10, 10)])
.await
.unwrap();
assert_no_drift(&writer, &format!("put {i}")).await;
}
assert!(
writer.memory().frozen_bytes() > 0
|| writer.memtable_stats().await.unwrap().frozen_count > 0,
"the loop must have sealed at least once for this to cover freeze"
);
writer.wait_for_flush_drain().await.unwrap();
assert_no_drift(&writer, "flush drain").await;
assert_eq!(
writer.memory().frozen_bytes(),
0,
"flushed memtables are reclaimable and must stop metering (grace {grace:?})"
);
for i in 24..32 {
writer
.put(vec![create_test_batch(&schema, i * 10, 10)])
.await
.unwrap();
assert_no_drift(&writer, &format!("post-flush put {i}")).await;
}
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_memory_is_readable_while_a_writer_holds_the_lock() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
..Default::default()
},
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 100)])
.await
.unwrap();
let unlocked = writer.memory().unflushed_bytes();
assert!(unlocked > 0, "rows are resident, so this cannot be zero");
let state_lock = writer.memtable_state_lock().unwrap().clone();
let held = state_lock.write().await;
assert_eq!(
writer.memory().unflushed_bytes(),
unlocked,
"a writer holding the lock must not zero the memory view"
);
assert!(
matches!(writer.memory().drain(), Drain::Stalled),
"the drain classification is published too, so it reads under the lock"
);
drop(held);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_both_seal_predicates_share_one_byte_arm() {
let schema = create_test_schema();
let mut memtable =
MemTable::with_capacity(schema.clone(), 1, vec![], CacheConfig::default(), 64).unwrap();
memtable
.insert(create_test_batch(&schema, 0, 50))
.await
.unwrap();
let at = memtable.batch_store().row_bytes();
assert!(at > 0, "the memtable must hold something to be a test");
for (bytes, expected) in [(at, true), (at + 1, false)] {
assert_eq!(
memtable.should_flush(bytes),
expected,
"should_flush at {bytes}"
);
assert_eq!(
memtable_reached_flush_threshold(&memtable, bytes, usize::MAX, usize::MAX, 1, 1),
expected,
"the two seal predicates disagree at {bytes}; a bloom-sized offset \
between them makes every memtable seal early on one path"
);
}
}
#[tokio::test]
async fn test_memory_handle_tracks_the_live_memtable() {
let schema = create_test_schema();
let mut memtable =
MemTable::with_capacity(schema.clone(), 1, vec![], CacheConfig::default(), 8).unwrap();
let handle = in_memory_ref(&memtable);
let empty_rows = handle.row_bytes();
for _ in 0..3 {
memtable
.insert(create_test_batch(&schema, 0, 10))
.await
.unwrap();
}
assert!(
handle.row_bytes() > empty_rows,
"a handle taken before the writes must still see them"
);
assert_eq!(
handle.row_bytes(),
in_memory_ref(&memtable).row_bytes(),
"a handle and a freshly taken one must agree"
);
assert_eq!(
handle.row_bytes(),
memtable.batch_store().row_bytes(),
"row bytes are the flush unit: batches only"
);
assert_eq!(
handle.index_bytes(),
super::super::memtable::pk_bloom_filter_bytes(),
"an unindexed memtable still holds its PK bloom filter"
);
assert_eq!(
handle.resident_bytes(),
handle.retained_row_bytes() + handle.index_bytes()
);
}
fn fake_memory(read: impl Fn() -> usize + Send + Sync + 'static) -> ShardMemory {
ShardMemory(ShardMemorySource::Fake(Arc::new(read)))
}
fn fixed_memory(unflushed: usize) -> ShardMemory {
fake_memory(move || unflushed)
}
fn empty_shard_memory() -> ShardMemory {
fixed_memory(0)
}
#[tokio::test]
async fn test_no_backpressure_when_under_threshold() {
let config = ShardWriterConfig::default().with_max_unflushed_memtable_bytes(1024 * 1024);
let controller = LocalBackpressureController::new(&config);
controller
.maybe_apply_backpressure(fixed_memory(100))
.await
.unwrap();
assert_eq!(controller.stats().count(), 0);
}
#[tokio::test]
async fn test_backpressure_loops_until_under_threshold() {
use std::sync::atomic::AtomicUsize;
use std::time::Duration;
let config = ShardWriterConfig::default()
.with_max_unflushed_memtable_bytes(100) .with_backpressure_log_interval(Duration::from_millis(50));
let call_count = Arc::new(AtomicUsize::new(0));
let call_count_clone = call_count.clone();
let controller = LocalBackpressureController::new(&config);
let draining = fake_memory(move || {
let count = call_count_clone.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1000usize.saturating_sub(count * 400)
});
controller.maybe_apply_backpressure(draining).await.unwrap();
assert_eq!(call_count.load(std::sync::atomic::Ordering::Relaxed), 4);
assert_eq!(controller.stats().count(), 1);
assert_eq!(
controller.stats().snapshot().active_count,
0,
"the wait is over, so nobody is parked"
);
}
#[tokio::test]
async fn test_backpressure_in_progress_wait_is_observable() {
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering as AtomicOrdering;
use std::time::Duration;
let config = ShardWriterConfig::default()
.with_max_unflushed_memtable_bytes(100)
.with_backpressure_log_interval(Duration::from_millis(50));
let unflushed = Arc::new(AtomicUsize::new(1000));
let release = unflushed.clone();
let controller = LocalBackpressureController::new(&config);
let stats = controller.stats().clone();
let parked = controller
.maybe_apply_backpressure(fake_memory(move || unflushed.load(AtomicOrdering::Relaxed)));
let observer = async {
let deadline = Instant::now() + Duration::from_secs(5);
while stats.snapshot().active_count == 0 {
assert!(
Instant::now() < deadline,
"an ongoing wait was never published"
);
tokio::time::sleep(Duration::from_millis(5)).await;
}
let mid = stats.snapshot();
assert_eq!(mid.active_count, 1);
assert_eq!(
mid.total_count, 0,
"a wait still in progress is not a completed one"
);
assert_eq!(mid.total_wait_ms, 0);
release.store(0, AtomicOrdering::Relaxed);
};
let (result, ()) = tokio::join!(parked, observer);
result.unwrap();
let after = stats.snapshot();
assert_eq!(after.active_count, 0, "the guard drops when the wait ends");
assert_eq!(after.total_count, 1);
}
#[tokio::test]
async fn test_backpressure_cancelled_wait_does_not_leak_active_count() {
use std::time::Duration;
let config = ShardWriterConfig::default()
.with_max_unflushed_memtable_bytes(100)
.with_backpressure_log_interval(Duration::from_millis(50));
let controller = LocalBackpressureController::new(&config);
assert!(
tokio::time::timeout(
Duration::from_millis(50),
controller.maybe_apply_backpressure(fixed_memory(1000)),
)
.await
.is_err()
);
let after = controller.stats().snapshot();
assert_eq!(after.active_count, 0, "cancellation must release the guard");
assert_eq!(
after.total_count, 0,
"a cancelled wait never completed, so it is not a completed wait"
);
}
#[derive(Debug)]
struct SpyController {
seen: Arc<StdRwLock<Vec<(usize, usize)>>>,
reject: bool,
}
#[async_trait::async_trait]
impl BackpressureController for SpyController {
async fn maybe_apply_backpressure(&self, shard: ShardMemory) -> Result<()> {
self.seen
.write()
.unwrap()
.push((shard.active_bytes(), shard.frozen_bytes()));
if self.reject {
return Err(Error::backpressure("full"));
}
Ok(())
}
}
#[tokio::test]
async fn test_injected_controller_replaces_default_and_sees_shard_memory() {
let seen = Arc::new(StdRwLock::new(Vec::new()));
let spy = Arc::new(SpyController {
seen: seen.clone(),
reject: false,
});
let config = ShardWriterConfig::default()
.with_max_unflushed_memtable_bytes(1)
.with_backpressure(spy);
let controller = resolve_backpressure(&config);
controller
.maybe_apply_backpressure(fixed_memory(1000))
.await
.unwrap();
assert_eq!(
*seen.read().unwrap(),
vec![(1000, 0)],
"the injected controller ran, and the built-in valve did not park the write"
);
}
#[tokio::test]
async fn test_shard_memory_reflects_drain_across_polls() {
#[derive(Debug)]
struct DrainWaiter {
polls: AtomicUsize,
}
#[async_trait::async_trait]
impl BackpressureController for DrainWaiter {
async fn maybe_apply_backpressure(&self, shard: ShardMemory) -> Result<()> {
while shard.unflushed_bytes() > 0 {
self.polls.fetch_add(1, Ordering::Relaxed);
tokio::task::yield_now().await;
}
Ok(())
}
}
let resident = Arc::new(AtomicUsize::new(4_096));
let drain = resident.clone();
let view = fake_memory(move || resident.load(Ordering::Relaxed));
let controller = Arc::new(DrainWaiter {
polls: AtomicUsize::new(0),
});
let gate = controller.clone();
let waiting = tokio::spawn(async move { gate.maybe_apply_backpressure(view).await });
tokio::task::yield_now().await;
drain.store(0, Ordering::Relaxed);
waiting.await.unwrap().unwrap();
assert!(controller.polls.load(Ordering::Relaxed) > 0);
}
#[tokio::test]
async fn test_default_backpressure_is_used_when_none_injected() {
let config = ShardWriterConfig::default().with_max_unflushed_memtable_bytes(100);
let controller = resolve_backpressure(&config);
let polls = Arc::new(AtomicUsize::new(0));
let polls_clone = polls.clone();
controller
.maybe_apply_backpressure(fake_memory(move || {
polls_clone.fetch_add(1, Ordering::Relaxed);
0
}))
.await
.unwrap();
assert_eq!(polls.load(Ordering::Relaxed), 1);
}
#[tokio::test]
async fn test_injected_controller_rejects_with_backpressure_error() {
let spy = Arc::new(SpyController {
seen: Arc::new(StdRwLock::new(Vec::new())),
reject: true,
});
let config = ShardWriterConfig::default().with_backpressure(spy);
let controller = resolve_backpressure(&config);
let err = controller
.maybe_apply_backpressure(empty_shard_memory())
.await
.unwrap_err();
assert!(err.is_backpressure(), "expected backpressure, got {err:?}");
}
#[test]
fn test_record_put() {
let stats = WriteStats::new();
stats.record_put(Duration::from_millis(10));
stats.record_put(Duration::from_millis(20));
let snapshot = stats.snapshot();
assert_eq!(snapshot.put_count, 2);
assert_eq!(snapshot.put_time, Duration::from_millis(30));
assert_eq!(snapshot.avg_put_latency(), Some(Duration::from_millis(15)));
}
#[test]
fn test_record_wal_flush() {
let stats = WriteStats::new();
stats.record_wal_flush(Duration::from_millis(100), 1024);
stats.record_wal_flush(Duration::from_millis(200), 2048);
let snapshot = stats.snapshot();
assert_eq!(snapshot.wal_flush_count, 2);
assert_eq!(snapshot.wal_flush_time, Duration::from_millis(300));
assert_eq!(snapshot.wal_flush_bytes, 3072);
assert_eq!(snapshot.avg_wal_flush_bytes(), Some(1536));
}
#[test]
fn test_record_memtable_flush() {
let stats = WriteStats::new();
stats.record_memtable_flush(Duration::from_secs(1), 10000);
let snapshot = stats.snapshot();
assert_eq!(snapshot.memtable_flush_count, 1);
assert_eq!(snapshot.memtable_flush_time, Duration::from_secs(1));
assert_eq!(snapshot.memtable_flush_rows, 10000);
}
#[test]
fn test_stats_reset() {
let stats = WriteStats::new();
stats.record_put(Duration::from_millis(10));
stats.record_wal_flush(Duration::from_millis(100), 1024);
stats.reset();
let snapshot = stats.snapshot();
assert_eq!(snapshot.put_count, 0);
assert_eq!(snapshot.wal_flush_count, 0);
}
fn wal_only_config(shard_id: Uuid) -> ShardWriterConfig {
ShardWriterConfig {
shard_id,
shard_spec_id: 0,
durable_write: true,
enable_memtable: false,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
manifest_scan_batch_size: 2,
..Default::default()
}
}
#[tokio::test]
async fn test_wal_only_durable_round_trip() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
let r1 = writer
.put(vec![create_test_batch(&schema, 0, 4)])
.await
.unwrap();
let r2 = writer
.put(vec![create_test_batch(&schema, 100, 2)])
.await
.unwrap();
assert_eq!(r1.batch_positions, 0..1);
assert_eq!(r2.batch_positions, 1..2);
writer.close().await.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
assert_eq!(tailer.first_position().await.unwrap(), 1);
assert_eq!(tailer.next_position().await.unwrap(), 3);
let e0 = tailer.read_entry(1).await.unwrap().unwrap();
let e1 = tailer.read_entry(2).await.unwrap().unwrap();
assert_eq!(e0.batches.len(), 1);
assert_eq!(e0.batches[0].num_rows(), 4);
assert_eq!(e1.batches.len(), 1);
assert_eq!(e1.batches[0].num_rows(), 2);
assert_eq!(e0.writer_epoch, e1.writer_epoch);
assert!(e0.writer_epoch >= 1);
}
#[tokio::test]
async fn test_wal_only_durable_put_waits_for_ticker_append() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let mut config = wal_only_config(shard_id);
config.max_wal_flush_interval = Some(Duration::from_millis(100));
config.max_wal_buffer_size = 100 * 1024 * 1024;
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config,
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 4)])
.await
.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
assert_eq!(tailer.next_position().await.unwrap(), 2);
let entry = tailer.read_entry(1).await.unwrap().unwrap();
assert_eq!(entry.batches.len(), 1);
assert_eq!(entry.batches[0].num_rows(), 4);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_rejects_index_configs() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let index_configs = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "id_idx".to_string(),
field_id: 0,
column: "id".to_string(),
})];
let err = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(Uuid::new_v4()),
schema,
index_configs,
)
.await
.err()
.expect("expected invalid_input");
assert!(
err.to_string().contains("indexes require enable_memtable"),
"unexpected error: {err}"
);
}
#[tokio::test]
async fn test_wal_only_rejects_empty_batches() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(Uuid::new_v4()),
schema.clone(),
vec![],
)
.await
.unwrap();
let err = writer.put(vec![]).await.err().unwrap();
assert!(err.to_string().contains("empty batch list"));
let zero = arrow_array::RecordBatch::new_empty(schema);
let err = writer.put(vec![zero]).await.err().unwrap();
assert!(err.to_string().contains("Batch 0 is empty"));
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_memtable_accessors_error() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(Uuid::new_v4()),
schema,
vec![],
)
.await
.unwrap();
let err = writer.memtable_stats().await.err().unwrap();
assert!(err.to_string().contains("WAL-only mode"));
let err = writer.scan().await.err().unwrap();
assert!(err.to_string().contains("WAL-only mode"));
let err = writer.active_memtable_ref().await.err().unwrap();
assert!(err.to_string().contains("WAL-only mode"));
let err = writer.in_memory_memtable_refs().await.err().unwrap();
assert!(err.to_string().contains("WAL-only mode"));
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_async_batches_multiple_puts() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let mut config = wal_only_config(Uuid::new_v4());
let shard_id = config.shard_id;
config.durable_write = false;
config.max_wal_flush_interval = None;
config.max_wal_buffer_size = 100 * 1024 * 1024;
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config,
schema.clone(),
vec![],
)
.await
.unwrap();
for i in 0..3 {
writer
.put(vec![create_test_batch(&schema, i * 10, 10)])
.await
.unwrap();
}
writer.close().await.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
assert_eq!(tailer.next_position().await.unwrap(), 2);
let entry = tailer.read_entry(1).await.unwrap().unwrap();
assert_eq!(entry.batches.len(), 3);
for (i, batch) in entry.batches.iter().enumerate() {
assert_eq!(batch.num_rows(), 10, "batch {i}");
}
}
#[tokio::test]
async fn test_wal_only_fencing() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
assert!(writer_b.epoch() > writer_a.epoch());
writer_b
.put(vec![create_test_batch(&schema, 1, 1)])
.await
.unwrap();
let err = writer_a
.put(vec![create_test_batch(&schema, 2, 1)])
.await
.expect_err("expected fence error");
assert!(
err.to_string().contains("Writer fenced"),
"unexpected error: {err}"
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_check_fenced_detects_successor_claim() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a.check_fenced().await.unwrap();
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
assert!(writer_b.epoch() > writer_a.epoch());
let err = writer_a
.check_fenced()
.await
.expect_err("expected fence error");
assert!(
err.to_string().contains("Writer fenced"),
"unexpected error: {err}"
);
writer_b.check_fenced().await.unwrap();
writer_b.close().await.unwrap();
}
fn memtable_config_with_pk(shard_id: Uuid) -> ShardWriterConfig {
ShardWriterConfig {
shard_id,
shard_spec_id: 0,
durable_write: true,
max_wal_buffer_size: 1024 * 1024,
max_wal_flush_interval: Some(Duration::from_millis(10)),
max_memtable_size: 64 * 1024 * 1024,
manifest_scan_batch_size: 2,
..Default::default()
}
}
fn schema_with_pk() -> Arc<ArrowSchema> {
use arrow_schema::Field;
let pk_meta: std::collections::HashMap<String, String> = [(
"lance-schema:unenforced-primary-key".to_string(),
"1".to_string(),
)]
.into_iter()
.collect();
let id_field = Field::new("id", DataType::Int32, false).with_metadata(pk_meta);
Arc::new(ArrowSchema::new(vec![
id_field,
Field::new("name", DataType::Utf8, true),
]))
}
#[tokio::test]
async fn test_writer_poisons_on_persistence_failure_and_recovers_on_reopen() {
let (store, base_path, controls) = failing_memory_store().await;
let base_uri = "memory:///";
let shard_id = Uuid::new_v4();
let schema = schema_with_pk();
controls.fail_wal_puts(usize::MAX);
let config = ShardWriterConfig {
max_wal_persist_retries: 1,
wal_persist_retry_base_delay: Duration::from_millis(1),
..memtable_config_with_pk(shard_id)
};
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config.clone(),
schema.clone(),
vec![],
)
.await
.unwrap();
let err = writer
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap_err();
assert_eq!(err.fence_reason(), Some(FenceReason::PersistenceFailure));
let err = writer
.put(vec![create_test_batch(&schema, 1, 1)])
.await
.unwrap_err();
assert_eq!(err.fence_reason(), Some(FenceReason::PersistenceFailure));
for reason in [
writer.scan().await.err().and_then(|e| e.fence_reason()),
writer
.active_memtable_ref()
.await
.err()
.and_then(|e| e.fence_reason()),
writer
.in_memory_memtable_refs()
.await
.err()
.and_then(|e| e.fence_reason()),
] {
assert_eq!(reason, Some(FenceReason::PersistenceFailure));
}
writer.memtable_stats().await.unwrap();
drop(writer);
controls.recover();
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 2, 1)])
.await
.unwrap();
}
#[tokio::test]
async fn test_doomed_open_does_not_fence_incumbent() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
let bad_fts = MemIndexConfig::Fts(FtsIndexConfig::new(
"bad_fts".to_string(),
0,
"id".to_string(),
));
let err = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![bad_fts],
)
.await
.map(|_| ())
.expect_err("an FTS index on a non-Utf8 column must be rejected");
assert!(
err.to_string().contains("bad_fts") && err.to_string().contains("Utf8"),
"unexpected error: {err}"
);
writer_a.check_fenced().await.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 1, 1)])
.await
.unwrap();
writer_a.close().await.unwrap();
}
#[tokio::test]
async fn test_freeze_dispatch_failure_retains_rows_and_poisons() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
durable_write: false,
max_wal_flush_interval: None,
..memtable_config_with_pk(shard_id)
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
assert_eq!(writer.memtable_stats().await.unwrap().row_count, 10);
writer.abort().await.unwrap();
let err = writer
.force_seal_active()
.await
.expect_err("force_seal_active must surface the failed dispatch");
assert!(
err.to_string().contains("channel closed"),
"unexpected error: {err}"
);
assert!(
writer.scan().await.is_err(),
"a poisoned writer must reject reads, not serve a divergent snapshot"
);
assert!(writer.in_memory_memtable_refs().await.is_err());
}
#[tokio::test]
async fn test_replay_rotates_when_wal_exceeds_one_memtable() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
const N: i32 = 8;
let writer_a_config = ShardWriterConfig {
max_memtable_batches: 1000,
..memtable_config_with_pk(shard_id)
};
let config = ShardWriterConfig {
max_memtable_batches: 2,
..memtable_config_with_pk(shard_id)
};
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
writer_a_config,
schema.clone(),
vec![],
)
.await
.unwrap();
for id in 0..N {
writer_a
.put(vec![create_test_batch(&schema, id, 1)])
.await
.unwrap();
}
}
async fn total_rows(writer: &ShardWriter, base_uri: &str, shard_id: Uuid) -> usize {
let mut rows = writer.memtable_stats().await.unwrap().row_count;
let manifest = writer.manifest().await.unwrap().unwrap();
for sstable in &manifest.sstables {
let gen_uri = format!("{}/_mem_wal/{}/{}", base_uri, shard_id, sstable.path);
let dataset = crate::Dataset::open(&gen_uri).await.unwrap();
rows += dataset.count_rows(None).await.unwrap();
}
rows
}
let writer_b = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
config.clone(),
schema.clone(),
vec![],
)
.await
.expect("a WAL larger than one memtable must still reopen");
let manifest = writer_b.manifest().await.unwrap().unwrap();
assert!(
!manifest.sstables.is_empty(),
"replay must have sealed and flushed at least one full memtable"
);
assert_eq!(
total_rows(&writer_b, &base_uri, shard_id).await as i32,
N,
"every replayed row must be durable, across generations and the active memtable"
);
writer_b.close().await.unwrap();
let writer_c =
ShardWriter::open(store, base_path, base_uri.clone(), config, schema, vec![])
.await
.unwrap();
assert_eq!(total_rows(&writer_c, &base_uri, shard_id).await as i32, N);
writer_c.close().await.unwrap();
}
#[tokio::test]
async fn test_replay_rotates_when_wal_exceeds_the_row_cap() {
use lance_arrow::FixedSizeListArrayExt;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let dim = 8;
let schema = hnsw_schema(dim);
let shard_id = Uuid::new_v4();
let cap = 8;
let vector_batch = |start: i32, rows: usize| {
let vectors = FixedSizeListArray::try_new_from_values(
Float32Array::from(
(0..rows * dim as usize)
.map(|v| v as f32 * 0.01)
.collect::<Vec<_>>(),
),
dim,
)
.unwrap();
RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(
(start..start + rows as i32).collect::<Vec<_>>(),
)),
Arc::new(vectors),
],
)
.unwrap()
};
let writer_a_config = ShardWriterConfig {
max_memtable_rows: 10_000,
..memtable_config_with_pk(shard_id)
};
let config = ShardWriterConfig {
max_memtable_rows: cap,
..memtable_config_with_pk(shard_id)
};
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
writer_a_config,
schema.clone(),
hnsw_configs(),
)
.await
.unwrap();
for round in 0..8i32 {
writer_a
.put(vec![vector_batch(round * 4, 4)])
.await
.unwrap();
}
}
let writer_b =
ShardWriter::open(store, base_path, base_uri, config, schema, hnsw_configs())
.await
.expect("a WAL holding more rows than the cap must still reopen");
let manifest = writer_b.manifest().await.unwrap().unwrap();
assert!(
!manifest.sstables.is_empty(),
"replay must have sealed and flushed the memtables it filled"
);
let stats = writer_b.memtable_stats().await.unwrap();
assert!(
stats.row_count <= cap,
"replay left {} rows in a memtable capped at {cap}",
stats.row_count
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_memtable_replay_recovers_unflushed_writes() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 100, 3)])
.await
.unwrap();
}
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![],
)
.await
.unwrap();
let stats = writer_b.memtable_stats().await.unwrap();
assert_eq!(
stats.row_count, 8,
"expected replay to insert 5 + 3 = 8 rows, got {}",
stats.row_count
);
assert_eq!(
stats.batch_count, 2,
"expected replay to insert 2 batches, got {}",
stats.batch_count
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_replay_does_not_reappend_or_reindex() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 100, 3)])
.await
.unwrap();
}
let writer_b = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
let stats = writer_b.memtable_stats().await.unwrap();
assert_eq!(stats.batch_count, 2);
assert_eq!(
stats.durable_batch_count, 2,
"replayed batches came from the WAL, so the durability cursor must already cover them"
);
writer_b
.put(vec![create_test_batch(&schema, 200, 2)])
.await
.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
let first = tailer.first_position().await.unwrap();
let next = tailer.next_position().await.unwrap();
let mut wal_rows = 0;
for position in first..next {
if let Some(entry) = tailer.read_entry(position).await.unwrap() {
wal_rows += entry.batches.iter().map(|b| b.num_rows()).sum::<usize>();
}
}
assert_eq!(
wal_rows, 10,
"WAL must hold 8 replayed + 2 new rows; a re-covering flush re-appends the replayed 8"
);
let mut scanner = writer_b.scan().await.unwrap();
scanner.filter("id = 0").unwrap();
let hit = scanner.try_into_batch().await.unwrap();
assert_eq!(
hit.num_rows(),
1,
"indexed PK lookup returned the replayed row more than once"
);
let all = writer_b
.scan()
.await
.unwrap()
.try_into_batch()
.await
.unwrap();
assert_eq!(all.num_rows(), 10);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_non_durable_put_is_visible_through_the_index() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
durable_write: false,
..memtable_config_with_pk(shard_id)
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
let mut scanner = writer.scan().await.unwrap();
scanner.filter("id = 3").unwrap();
let hit = scanner.try_into_batch().await.unwrap();
assert_eq!(
hit.num_rows(),
1,
"an index-backed lookup must see a non-durable put as soon as it returns"
);
let all = writer.scan().await.unwrap().try_into_batch().await.unwrap();
assert_eq!(all.num_rows(), 5);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_durable_ack_after_rotation_requires_its_own_wal_append() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
max_memtable_batches: 2,
..memtable_config_with_pk(shard_id)
};
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config,
schema.clone(),
vec![],
)
.await
.unwrap();
for i in 0..3 {
writer
.put(vec![create_test_batch(&schema, i * 5, 5)])
.await
.unwrap();
}
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.global_offset > 0,
"expected a rotation; the active memtable is still the writer's first"
);
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.durable_batch_count >= stats.global_offset + stats.batch_count,
"durable_write acked a put the WAL never received: durable={} but the active \
memtable spans [{}, {})",
stats.durable_batch_count,
stats.global_offset,
stats.global_offset + stats.batch_count
);
writer.close().await.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
let first = tailer.first_position().await.unwrap();
let next = tailer.next_position().await.unwrap();
let mut wal_rows = 0;
for position in first..next {
if let Some(entry) = tailer.read_entry(position).await.unwrap() {
wal_rows += entry.batches.iter().map(|b| b.num_rows()).sum::<usize>();
}
}
assert_eq!(
wal_rows, 15,
"every acked row must be in the WAL; a post-rotation put that acked without an \
append would leave rows missing"
);
}
#[test]
fn test_next_pending_store_picks_the_oldest_owing_an_append() {
let schema = create_test_schema();
let frozen = Arc::new(BatchStore::with_capacity(4));
frozen.append(create_test_batch(&schema, 0, 1)).unwrap();
frozen.append(create_test_batch(&schema, 1, 1)).unwrap();
let active = Arc::new(BatchStore::with_capacity_at(4, 2));
active.append(create_test_batch(&schema, 2, 1)).unwrap();
let frozen_list = || std::iter::once(Arc::clone(&frozen));
let picked = next_pending_store(frozen_list(), Arc::clone(&active), 0).unwrap();
assert!(
Arc::ptr_eq(&picked, &frozen),
"the outgoing memtable's tail must be appended before the incoming one's head"
);
let picked = next_pending_store(frozen_list(), Arc::clone(&active), 1).unwrap();
assert!(Arc::ptr_eq(&picked, &frozen));
let picked = next_pending_store(frozen_list(), Arc::clone(&active), 2).unwrap();
assert!(Arc::ptr_eq(&picked, &active));
assert!(next_pending_store(frozen_list(), Arc::clone(&active), 3).is_none());
}
#[rstest]
#[case::memtable(true)]
#[case::wal_only(false)]
#[tokio::test]
async fn test_open_rejects_durable_write_without_a_ticker(#[case] enable_memtable: bool) {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
durable_write: true,
enable_memtable,
max_wal_flush_interval: None,
..memtable_config_with_pk(shard_id)
};
let Err(err) = ShardWriter::open(store, base_path, base_uri, config, schema, vec![]).await
else {
panic!("durable_write with no ticker must be rejected");
};
assert!(
err.to_string().contains("max_wal_flush_interval"),
"the error must name the knob, got: {err}"
);
}
#[tokio::test]
async fn test_wal_append_order_preserves_pk_recency_across_rotation() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig {
max_memtable_batches: 2,
..memtable_config_with_pk(shard_id)
};
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config,
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 7, 1)])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 20, 1)])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 30, 1)])
.await
.unwrap();
writer.close().await.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
let first = tailer.first_position().await.unwrap();
let next = tailer.next_position().await.unwrap();
let mut ids: Vec<i32> = Vec::new();
for position in first..next {
let Some(entry) = tailer.read_entry(position).await.unwrap() else {
continue;
};
for batch in &entry.batches {
let column = batch
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int32Array>()
.unwrap();
ids.extend((0..column.len()).map(|i| column.value(i)));
}
}
assert_eq!(
ids,
vec![7, 20, 30],
"WAL entries must follow global batch-position order; memtable 1's rows must \
precede memtable 2's, or replay inverts primary-key recency"
);
}
#[tokio::test]
async fn test_open_rejects_index_config_that_disagrees_with_schema() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let bad_fts = MemIndexConfig::Fts(FtsIndexConfig::new(
"bad_fts".to_string(),
0,
"id".to_string(),
));
let Err(err) = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![bad_fts],
)
.await
else {
panic!("open must reject an FTS index on a non-Utf8 column");
};
let message = err.to_string();
assert!(
message.contains("bad_fts") && message.contains("Utf8"),
"error must name the index and the constraint, got: {message}"
);
}
#[tokio::test]
async fn test_memtable_replay_no_op_on_fresh_shard() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![],
)
.await
.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.row_count, 0);
assert_eq!(stats.batch_count, 0);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_memtable_replay_skips_entries_after_external_compaction() {
use crate::dataset::mem_wal::ShardManifestStore;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 5)])
.await
.unwrap();
writer_a.close().await.unwrap();
}
let manifest_store = ShardManifestStore::new(store.clone(), &base_path, shard_id, 2);
let pre = manifest_store.latest().await.unwrap().unwrap();
assert!(
!pre.sstables.is_empty(),
"writer A's close() should have stamped an SSTable"
);
let cursor_at_flush = pre.replay_after_wal_entry_position;
assert!(
cursor_at_flush >= 1,
"expected cursor to land on a 1-based WAL position after flush, got {cursor_at_flush}"
);
let (compactor_epoch, _) = manifest_store.claim_epoch(pre.shard_spec_id).await.unwrap();
manifest_store
.commit_update(compactor_epoch, |current| ShardManifest {
version: current.next_version(),
sstables: vec![],
..current.clone()
})
.await
.unwrap();
let post = manifest_store.latest().await.unwrap().unwrap();
assert!(
post.sstables.is_empty(),
"compactor drain should have left sstables empty"
);
assert_eq!(
post.replay_after_wal_entry_position, cursor_at_flush,
"compactor must not touch the replay cursor"
);
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![],
)
.await
.unwrap();
let stats = writer_b.memtable_stats().await.unwrap();
assert_eq!(
stats.row_count, 0,
"memtable must not re-replay compacted WAL entries; got {} rows",
stats.row_count
);
assert_eq!(stats.batch_count, 0);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_memtable_replay_fenced_aborts_open() {
use crate::dataset::mem_wal::ShardManifestStore;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
{
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
}
let manifest_store = Arc::new(ShardManifestStore::new(
store.clone(),
&base_path,
shard_id,
2,
));
let high_epoch_appender = WalAppender::with_claimed_epoch(
store.clone(),
base_path.clone(),
shard_id,
manifest_store,
100,
0,
WalRetryConfig::default(),
);
high_epoch_appender
.append(vec![create_test_batch(&schema, 999, 1)])
.await
.unwrap();
let result = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![],
)
.await;
let Err(err) = result else {
panic!("expected open to fail with fence error during replay");
};
assert_eq!(
err.fence_reason(),
Some(FenceReason::PeerClaimedEpoch),
"replay must abort with a typed peer-fence error, got: {err}"
);
let msg = err.to_string();
assert!(
msg.contains("WAL replay aborted") && msg.contains("fenced"),
"unexpected error: {msg}"
);
}
#[tokio::test]
async fn test_wal_stats_seeded_from_manifest_on_reopen() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer1 = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer1
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
writer1.close().await.unwrap();
let writer2 = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(shard_id),
schema,
vec![],
)
.await
.unwrap();
let next = writer2.wal_stats().next_wal_entry_position;
assert!(
next >= 1,
"expected wal_stats to reflect post-recovery cursor (>= 1) on reopen, got {next}"
);
writer2.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_size_trigger_fires_repeatedly() {
use crate::dataset::mem_wal::wal::WalTailer;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let mut config = wal_only_config(Uuid::new_v4());
let shard_id = config.shard_id;
config.durable_write = false;
config.max_wal_flush_interval = None;
config.max_wal_buffer_size = 1;
let writer = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri,
config,
schema.clone(),
vec![],
)
.await
.unwrap();
for i in 0..3 {
writer
.put(vec![create_test_batch(&schema, i * 10, 10)])
.await
.unwrap();
tokio::task::yield_now().await;
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
writer.close().await.unwrap();
let tailer = WalTailer::new(store, base_path, shard_id);
let next = tailer.next_position().await.unwrap();
assert!(
next >= 3,
"expected at least 3 WAL entries (one per crossing), got next_position = {next}"
);
}
#[tokio::test]
async fn test_wal_only_fenced_concurrent_puts_do_not_silently_succeed() {
use std::sync::Arc;
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer_a = Arc::new(
ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap(),
);
writer_a
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_b
.put(vec![create_test_batch(&schema, 1, 1)])
.await
.unwrap();
let a1 = writer_a.clone();
let a2 = writer_a.clone();
let schema1 = schema.clone();
let schema2 = schema.clone();
let h1 = tokio::spawn(async move { a1.put(vec![create_test_batch(&schema1, 2, 1)]).await });
let h2 = tokio::spawn(async move { a2.put(vec![create_test_batch(&schema2, 3, 1)]).await });
let r1 = h1.await.unwrap();
let r2 = h2.await.unwrap();
assert!(
r1.is_err() && r2.is_err(),
"expected both concurrent puts on a fenced writer to fail, got r1={r1:?} r2={r2:?}",
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_wal_only_stats_no_memtable_flush() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
wal_only_config(Uuid::new_v4()),
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 1)])
.await
.unwrap();
let stats_handle = writer.stats_handle();
writer.close().await.unwrap();
let snapshot = stats_handle.snapshot();
assert!(snapshot.put_count >= 1, "expected at least one put");
assert!(
snapshot.wal_flush_count >= 1,
"expected at least one WAL flush"
);
assert_eq!(
snapshot.memtable_flush_count, 0,
"WAL-only mode must never trigger a memtable flush"
);
assert_eq!(
snapshot.index_update_count, 0,
"WAL-only mode must never trigger an index update"
);
}
#[tokio::test]
async fn test_memtable_stats_record_index_update() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_pk_test_schema();
let index_configs = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "id_idx".to_string(),
field_id: 0,
column: "id".to_string(),
})];
let writer = ShardWriter::open(
store,
base_path,
base_uri,
flush_test_config(Uuid::new_v4()),
schema.clone(),
index_configs,
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 3)])
.await
.unwrap();
let stats_handle = writer.stats_handle();
writer.close().await.unwrap();
let snapshot = stats_handle.snapshot();
assert!(
snapshot.index_update_count >= 1,
"the index apply must record an index-update stat, got {}",
snapshot.index_update_count
);
assert_eq!(
snapshot.index_update_rows, 3,
"every indexed row must be counted exactly once, got {}",
snapshot.index_update_rows
);
assert!(
snapshot.avg_index_update_latency().is_some(),
"a recorded index update must expose an average latency"
);
}
#[tokio::test]
async fn test_force_seal_active_and_wait_for_flush_drain() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
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 writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
let flushed_before = writer
.manifest()
.await
.unwrap()
.map(|m| m.sstables.len())
.unwrap_or(0);
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
let fence = writer.force_seal_active().await.unwrap();
assert_eq!(fence.sealed_generation(), Some(initial_gen));
fence.wait().await.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert_eq!(stats.generation, initial_gen + 1);
assert_eq!(stats.batch_count, 0);
let manifest = writer
.manifest()
.await
.unwrap()
.expect("manifest should exist after flush");
assert_eq!(manifest.sstables.len(), flushed_before + 1);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_observer_sees_both_flush_kinds() {
#[derive(Debug, Default)]
struct CountingObserver {
wal_flushes: AtomicU64,
wal_bytes: AtomicU64,
memtable_flushes: AtomicU64,
memtable_rows: AtomicU64,
}
impl WalObserver for CountingObserver {
fn on_wal_flush(&self, _duration: Duration, bytes: usize) {
self.wal_flushes.fetch_add(1, Ordering::Relaxed);
self.wal_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
}
fn on_memtable_flush(&self, _duration: Duration, rows: usize) {
self.memtable_flushes.fetch_add(1, Ordering::Relaxed);
self.memtable_rows.fetch_add(rows as u64, Ordering::Relaxed);
}
}
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let observer = Arc::new(CountingObserver::default());
let sink: Arc<dyn WalObserver> = observer.clone();
let config = ShardWriterConfig {
observer: Some(sink),
..seal_fence_test_config(Uuid::new_v4())
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer
.force_seal_active()
.await
.unwrap()
.wait()
.await
.unwrap();
assert!(observer.wal_flushes.load(Ordering::Relaxed) > 0);
assert!(observer.wal_bytes.load(Ordering::Relaxed) > 0);
assert_eq!(observer.memtable_flushes.load(Ordering::Relaxed), 1);
assert_eq!(observer.memtable_rows.load(Ordering::Relaxed), 10);
writer.close().await.unwrap();
}
fn seal_fence_test_config(shard_id: Uuid) -> ShardWriterConfig {
ShardWriterConfig {
shard_id,
shard_spec_id: 0,
durable_write: true,
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 test_force_seal_active_fences_pending_generation_when_active_is_empty() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
seal_fence_test_config(Uuid::new_v4()),
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer.task_executor.shutdown_all().await.unwrap();
let pending_generation = match &writer.mode {
WriterMode::MemTable {
state,
writer_state,
..
} => {
let mut state = state.write().await;
writer_state.freeze_memtable(&mut state).unwrap() - 1
}
WriterMode::WalOnly { .. } => unreachable!("opened in memtable mode"),
};
let fence = writer.force_seal_active().await.unwrap();
assert_eq!(
fence.sealed_generation(),
None,
"the active memtable was empty, so this seal froze nothing"
);
assert!(
tokio::time::timeout(Duration::from_millis(200), fence.wait())
.await
.is_err(),
"generation {pending_generation} is still awaiting flush; the fence must not be satisfied"
);
}
#[tokio::test]
async fn test_force_seal_active_fence_ignores_manifest_generation_advance() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let writer = ShardWriter::open(
store,
base_path,
base_uri,
seal_fence_test_config(Uuid::new_v4()),
schema.clone(),
vec![],
)
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer.task_executor.shutdown_all().await.unwrap();
let fence = writer.force_seal_active().await.unwrap();
let sealed = fence
.sealed_generation()
.expect("the active memtable held rows");
writer
.manifest_store
.commit_update(writer.epoch(), |current| ShardManifest {
version: current.next_version(),
current_generation: sealed + 2,
..current.clone()
})
.await
.unwrap();
assert!(
tokio::time::timeout(Duration::from_millis(200), fence.wait())
.await
.is_err(),
"generation {sealed} never reached L0; a manifest advance past it must not satisfy its fence"
);
}
#[tokio::test]
async fn test_abort_discards_without_flushing_and_is_idempotent() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
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 writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
let flushed_before = writer
.manifest()
.await
.unwrap()
.map(|m| m.sstables.len())
.unwrap_or(0);
writer.abort().await.unwrap();
let flushed_after = writer
.manifest()
.await
.unwrap()
.map(|m| m.sstables.len())
.unwrap_or(0);
assert_eq!(
flushed_after, flushed_before,
"abort must not flush a new L0 generation"
);
writer.abort().await.unwrap();
}
#[tokio::test]
async fn test_frozen_retained_during_grace_then_swept() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
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,
frozen_memtable_grace: Duration::from_millis(50),
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
assert!(
manifest
.sstables
.iter()
.any(|g| g.generation == initial_gen),
"SSTable must be recorded in the manifest"
);
let refs = writer.in_memory_memtable_refs().await.unwrap();
assert_eq!(refs.active.generation, initial_gen + 1);
assert!(
refs.frozen.iter().any(|f| f.generation == initial_gen),
"SSTable must stay queryable during the grace window"
);
tokio::time::sleep(Duration::from_millis(250)).await;
let refs = writer.in_memory_memtable_refs().await.unwrap();
assert!(
refs.frozen.is_empty(),
"frozen handle must be swept once the grace elapses"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_frozen_evicted_immediately_with_zero_grace() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
shard_spec_id: 0,
durable_write: false,
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,
frozen_memtable_grace: Duration::ZERO,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let initial_gen = writer.memtable_stats().await.unwrap().generation;
writer
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
assert!(
manifest
.sstables
.iter()
.any(|g| g.generation == initial_gen),
"SSTable must be recorded in the manifest"
);
let refs = writer.in_memory_memtable_refs().await.unwrap();
assert!(
refs.frozen.is_empty(),
"frozen handle must be evicted on commit when grace is zero"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_close_propagates_frozen_memtable_flush_failure() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = schema_with_pk();
let shard_id = Uuid::new_v4();
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
writer_a
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema,
vec![],
)
.await
.unwrap();
assert!(writer_b.epoch() > writer_a.epoch());
let error = writer_a
.close()
.await
.expect_err("close must propagate the fenced MemTable flush");
assert!(
matches!(error, Error::IO { .. }),
"unexpected error: {error}"
);
assert!(
error.to_string().contains("Writer fenced"),
"unexpected error: {error}"
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_frozen_retained_after_failed_flush() {
let (store, base_path, base_uri, _temp_dir) = create_local_store().await;
let schema = create_test_schema();
let shard_id = Uuid::new_v4();
let writer_a = ShardWriter::open(
store.clone(),
base_path.clone(),
base_uri.clone(),
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
let initial_gen = writer_a.memtable_stats().await.unwrap().generation;
writer_a
.put(vec![create_test_batch(&schema, 0, 10)])
.await
.unwrap();
let writer_b = ShardWriter::open(
store,
base_path,
base_uri,
memtable_config_with_pk(shard_id),
schema.clone(),
vec![],
)
.await
.unwrap();
assert!(writer_b.epoch() > writer_a.epoch());
match &writer_a.mode {
WriterMode::MemTable {
state,
writer_state,
..
} => {
let mut st = state.write().await;
writer_state.freeze_memtable(&mut st).unwrap();
}
WriterMode::WalOnly { .. } => unreachable!("opened in memtable mode"),
}
assert!(
writer_a.wait_for_flush_drain().await.is_err(),
"fenced flush should fail the drain"
);
let refs = writer_a.in_memory_memtable_refs().await.unwrap();
assert_eq!(refs.frozen.len(), 1, "sealed generation must be retained");
assert_eq!(refs.frozen[0].generation, initial_gen);
assert!(
!refs.frozen[0].batch_store.is_empty(),
"retained sealed memtable must still hold its rows"
);
assert_eq!(refs.active.generation, initial_gen + 1);
let stats = writer_a.memtable_stats().await.unwrap();
assert_eq!(stats.frozen_count, 1);
let frozen_bytes = writer_a.memory().frozen_bytes();
assert!(
frozen_bytes >= refs.frozen[0].batch_store.row_bytes(),
"a failed flush must keep owing its resident bytes, got {frozen_bytes}"
);
assert!(
matches!(writer_a.memory().drain(), Drain::Stalled),
"a failed flush leaves nothing outstanding to wait on"
);
let controller = LocalBackpressureController::new(&ShardWriterConfig {
max_unflushed_memtable_bytes: writer_a.memory().unflushed_bytes(),
..Default::default()
});
let refused = tokio::time::timeout(
STALL_GRACE * 10,
controller.maybe_apply_backpressure(writer_a.memory()),
)
.await
.expect("a shard nothing can drain must refuse, not park the writer");
assert!(
refused.is_err_and(|e| e.is_backpressure()),
"the refusal must be the retryable backpressure signal"
);
writer_b.close().await.unwrap();
}
#[tokio::test]
async fn test_pinned_parents_seal_before_the_ceiling_traps_the_writer() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let schema = Arc::new(ArrowSchema::new(vec![Field::new(
"id",
DataType::Int32,
false,
)]));
let chunk = 1_000_000; let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
max_memtable_size: 1024 * 1024,
max_memtable_batches: 1024,
max_unflushed_memtable_bytes: 8 * 1024 * 1024,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
for i in 0..6i32 {
let parent = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int32Array::from(
(0..chunk).map(|v| v + i).collect::<Vec<_>>(),
))],
)
.unwrap();
tokio::time::timeout(STALL_GRACE * 10, writer.put(vec![parent.slice(0, 1)]))
.await
.expect("a shard the resident arm can seal must not park forever")
.unwrap_or_else(|e| {
panic!("put {i} was refused, so the ceiling is still a trap: {e}")
});
}
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.generation > 0,
"the pinned parents must have sealed a generation; row bytes never \
came close to max_memtable_size"
);
assert!(
writer.memory().row_bytes() < 1024,
"the row window must still be tiny — otherwise the row arm did the \
sealing and this proves nothing"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_resident_arm_seals_on_index_memory() {
let schema = create_test_schema();
let mut memtable =
MemTable::with_capacity(schema.clone(), 1, vec![], CacheConfig::default(), 64).unwrap();
memtable
.insert(create_test_batch(&schema, 0, 50))
.await
.unwrap();
let resident = memtable_resident_bytes(&memtable);
let rows = memtable.batch_store().row_bytes();
assert!(
resident > rows,
"the fixture needs non-row memory to be measuring anything: \
resident {resident} vs rows {rows}"
);
assert!(
memtable_reached_flush_threshold(&memtable, usize::MAX, usize::MAX, resident, 1, 1),
"resident bytes at the ceiling must seal"
);
assert!(
!memtable_reached_flush_threshold(
&memtable,
usize::MAX,
usize::MAX,
resident + 1,
1,
1
),
"and must not seal below it"
);
}
#[tokio::test]
async fn test_row_arm_seals_on_max_memtable_rows() {
let schema = create_test_schema();
let mut memtable =
MemTable::with_capacity(schema.clone(), 1, vec![], CacheConfig::default(), 64).unwrap();
memtable
.insert(create_test_batch(&schema, 0, 50))
.await
.unwrap();
let row_arm = |cap, incoming| {
memtable_reached_flush_threshold(&memtable, usize::MAX, cap, usize::MAX, 1, incoming)
};
assert!(!row_arm(50, 0), "50 rows under a cap of 50 must not seal");
assert!(row_arm(50, 1), "no room for one more row must seal");
assert!(
!row_arm(60, 10),
"a put that exactly fills the cap must not seal"
);
assert!(
row_arm(60, 11),
"a put that would overflow the cap must seal"
);
}
#[tokio::test]
async fn test_put_seals_on_max_memtable_rows() {
let (store, base_path, base_uri, _t) = create_local_store().await;
let schema = create_test_schema();
let cap = 64;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
max_memtable_rows: cap,
max_memtable_size: 64 * 1024 * 1024,
max_memtable_batches: 8_000,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
for round in 0..10i32 {
writer
.put(vec![create_test_batch(&schema, round * 10, 10)])
.await
.unwrap();
let stats = writer.memtable_stats().await.unwrap();
assert!(
stats.row_count <= cap,
"the active memtable holds {} rows, past the cap of {cap}",
stats.row_count
);
}
assert!(
writer.memtable_stats().await.unwrap().generation > 1,
"100 rows under a cap of {cap} must have rotated at least once"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_put_rejects_more_rows_than_a_memtable_holds() {
let (store, base_path, base_uri, _t) = create_local_store().await;
let schema = create_test_schema();
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
max_memtable_rows: 8,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), vec![])
.await
.unwrap();
let err = writer
.put(vec![
create_test_batch(&schema, 0, 5),
create_test_batch(&schema, 5, 4),
])
.await
.expect_err("a put of 9 rows under a cap of 8 must be rejected");
assert!(
matches!(err, Error::InvalidInput { .. }),
"an oversized put is caller error, not a writer fault: {err}"
);
assert!(
err.to_string().contains("max_memtable_rows=8"),
"the error must name the knob and its value, got: {err}"
);
writer
.put(vec![create_test_batch(&schema, 0, 8)])
.await
.unwrap();
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_hnsw_index_survives_a_shard_that_outgrows_the_row_cap() {
use lance_arrow::FixedSizeListArrayExt;
let (store, base_path, base_uri, _t) = create_local_store().await;
let dim = 8;
let schema = hnsw_schema(dim);
let cap = 64;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
durable_write: false,
max_memtable_rows: cap,
max_memtable_size: 64 * 1024 * 1024,
max_memtable_batches: 8_000,
..Default::default()
};
let writer = ShardWriter::open(
store,
base_path,
base_uri,
config,
schema.clone(),
hnsw_configs(),
)
.await
.unwrap();
for round in 0..20i32 {
let ids: Vec<i32> = (0..10).map(|v| v + round * 10).collect();
let vectors = FixedSizeListArray::try_new_from_values(
Float32Array::from(
(0..10 * dim as usize)
.map(|v| v as f32 * 0.01)
.collect::<Vec<_>>(),
),
dim,
)
.unwrap();
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int32Array::from(ids)), Arc::new(vectors)],
)
.unwrap();
writer
.put(vec![batch])
.await
.unwrap_or_else(|e| panic!("put {round} was refused: {e}"));
}
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_index_growth_after_the_seal_check_still_drains() {
let (store, base_path, base_uri, _t) = create_local_store().await;
let schema = Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new("text", DataType::Utf8, true),
]));
let btree = vec![MemIndexConfig::BTree(BTreeIndexConfig {
name: "text_idx".to_string(),
field_id: 1,
column: "text".to_string(),
})];
let rows = 2_000usize;
let width = 512usize;
let payload = rows * width;
let ceiling = payload + payload / 2;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
max_memtable_size: ceiling - 64 * 1024,
max_memtable_batches: 1024,
max_unflushed_memtable_bytes: ceiling,
..Default::default()
};
let writer = ShardWriter::open(store, base_path, base_uri, config, schema.clone(), btree)
.await
.unwrap();
for i in 0..3i32 {
let ids: Vec<i32> = (0..rows as i32).map(|v| v + i * rows as i32).collect();
let texts: Vec<String> = ids
.iter()
.map(|v| format!("{v:07}{}", "z".repeat(width - 7)))
.collect();
let batch = RecordBatch::try_new(
schema.clone(),
vec![
Arc::new(Int32Array::from(ids)),
Arc::new(StringArray::from(
texts.iter().map(|t| Some(t.as_str())).collect::<Vec<_>>(),
)),
],
)
.unwrap();
tokio::time::timeout(STALL_GRACE * 10, writer.put(vec![batch]))
.await
.expect("a shard with a sealable memtable must not park forever")
.unwrap_or_else(|e| {
panic!("put {i} was refused; index growth outran the seal check: {e}")
});
}
assert!(
writer.memory().index_bytes() > writer.memory().row_bytes(),
"the fixture must be index-dominated, or it is not testing this path"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_open_rejects_a_zero_row_headroom() {
let sizing = ShardWriterConfig {
max_memtable_rows: 2_000,
..Default::default()
};
let reserved = reserved_index_bytes(&sizing, &hnsw_configs());
let (store, base_path, base_uri, _t) = create_local_store().await;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
max_memtable_rows: sizing.max_memtable_rows,
max_memtable_size: 0,
max_unflushed_memtable_bytes: reserved,
..Default::default()
};
let err = ShardWriter::open(
store,
base_path,
base_uri,
config,
hnsw_schema(8),
hnsw_configs(),
)
.await
.err()
.expect("a zero row headroom must be rejected at open");
assert!(
err.to_string()
.contains("max_memtable_size must be greater than zero"),
"the error must name the knob, got: {err}"
);
}
fn hnsw_schema(dim: i32) -> Arc<ArrowSchema> {
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int32, false),
Field::new(
"vector",
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), dim),
true,
),
]))
}
fn hnsw_configs() -> Vec<MemIndexConfig> {
vec![MemIndexConfig::Hnsw(Box::new(HnswIndexConfig::new(
"vec_idx".to_string(),
1,
"vector".to_string(),
lance_linalg::distance::DistanceType::L2,
)))]
}
fn reserved_index_bytes(config: &ShardWriterConfig, configs: &[MemIndexConfig]) -> usize {
IndexStore::from_configs(
configs,
config.max_memtable_rows,
config.max_memtable_batches,
)
.unwrap()
.resident_bytes()
+ super::super::memtable::pk_bloom_filter_bytes()
}
#[tokio::test]
async fn test_open_rejects_indexes_that_cannot_fit_under_the_ceiling() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
max_memtable_rows: 100_000,
max_memtable_size: 1024 * 1024,
max_unflushed_memtable_bytes: 1024 * 1024,
..Default::default()
};
let reserved = reserved_index_bytes(&config, &hnsw_configs());
assert!(
reserved > config.max_unflushed_memtable_bytes,
"the fixture must actually over-subscribe the ceiling, got {reserved}"
);
let err = ShardWriter::open(
store,
base_path,
base_uri,
config,
hnsw_schema(32),
hnsw_configs(),
)
.await
.err()
.expect("an over-subscribed ceiling must be rejected at open");
let message = err.to_string();
for fragment in [
"in-memory indexes reserve",
"max_unflushed_memtable_bytes",
"stalling every write",
] {
assert!(
message.contains(fragment),
"the error must name {fragment}, got: {message}"
);
}
assert!(
!err.is_backpressure(),
"a config error must not masquerade as the retryable busy signal"
);
}
#[tokio::test]
async fn test_writer_with_indexes_under_the_ceiling_keeps_accepting_writes() {
let (store, base_path, base_uri, _temp) = create_local_store().await;
let dim = 8;
let sizing = ShardWriterConfig {
max_memtable_rows: 2_000,
..Default::default()
};
let reserved = reserved_index_bytes(&sizing, &hnsw_configs());
let max_memtable_size = 4 * 1024;
let config = ShardWriterConfig {
shard_id: Uuid::new_v4(),
max_memtable_rows: sizing.max_memtable_rows,
max_memtable_size,
max_unflushed_memtable_bytes: 2 * (reserved + max_memtable_size),
..Default::default()
};
let schema = hnsw_schema(dim);
let writer = ShardWriter::open(
store,
base_path,
base_uri,
config,
schema.clone(),
hnsw_configs(),
)
.await
.expect("a reservation that leaves room under the ceiling must open");
for round in 0..8i32 {
let rows = 64;
let ids: Vec<i32> = (0..rows).map(|i| round * rows + i).collect();
let values: Vec<f32> = (0..rows * dim).map(|i| i as f32).collect();
let vectors = FixedSizeListArray::try_new(
Arc::new(Field::new("item", DataType::Float32, true)),
dim,
Arc::new(Float32Array::from(values)),
None,
)
.unwrap();
let batch = RecordBatch::try_new(
schema.clone(),
vec![Arc::new(Int32Array::from(ids)), Arc::new(vectors)],
)
.unwrap();
tokio::time::timeout(STALL_GRACE * 10, writer.put(vec![batch]))
.await
.expect("a drainable shard must not park the writer indefinitely")
.unwrap_or_else(|e| panic!("put in round {round} was refused: {e}"));
}
assert!(
writer.memtable_stats().await.unwrap().generation > 0,
"the run must cross a seal for the drain path to have been exercised"
);
writer.close().await.unwrap();
}
}
#[cfg(test)]
mod shard_writer_tests {
use std::sync::Arc;
use crate::index::DatasetIndexExt;
use arrow_array::{
FixedSizeListArray, Float32Array, Int64Array, RecordBatch, RecordBatchIterator, StringArray,
};
use arrow_schema::{DataType, Field, Schema as ArrowSchema};
use lance_arrow::FixedSizeListArrayExt;
use lance_index::IndexType;
use lance_index::scalar::inverted::{InvertedIndexParams, InvertedListFormatVersion};
use lance_index::scalar::{FullTextSearchQuery, ScalarIndexParams};
use lance_index::vector::ivf::IvfBuildParams;
use lance_index::vector::pq::builder::PQBuildParams;
use lance_linalg::distance::MetricType;
use uuid::Uuid;
use crate::dataset::mem_wal::DatasetMemWalExt;
use crate::dataset::{Dataset, WriteParams};
use crate::index::vector::VectorIndexParams;
use super::super::ShardWriterConfig;
fn create_test_schema(vector_dim: i32) -> Arc<ArrowSchema> {
use std::collections::HashMap;
let mut id_metadata = HashMap::new();
id_metadata.insert(
"lance-schema:unenforced-primary-key".to_string(),
"true".to_string(),
);
let id_field = Field::new("id", DataType::Int64, false).with_metadata(id_metadata);
Arc::new(ArrowSchema::new(vec![
id_field,
Field::new(
"vector",
DataType::FixedSizeList(
Arc::new(Field::new("item", DataType::Float32, true)),
vector_dim,
),
true,
),
Field::new("text", DataType::Utf8, true),
]))
}
fn create_append_only_schema(vector_dim: i32) -> Arc<ArrowSchema> {
Arc::new(ArrowSchema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new(
"vector",
DataType::FixedSizeList(
Arc::new(Field::new("item", DataType::Float32, true)),
vector_dim,
),
true,
),
Field::new("text", DataType::Utf8, true),
]))
}
fn create_test_batch(
schema: &ArrowSchema,
start_id: i64,
num_rows: usize,
vector_dim: i32,
) -> RecordBatch {
let vectors: Vec<f32> = (0..num_rows)
.flat_map(|i| {
let seed = (start_id as usize + i) as f32;
(0..vector_dim as usize).map(move |d| (seed * 0.1 + d as f32 * 0.01).sin())
})
.collect();
let vector_array =
FixedSizeListArray::try_new_from_values(Float32Array::from(vectors), vector_dim)
.unwrap();
let texts: Vec<String> = (0..num_rows)
.map(|i| format!("Sample text for row {}", start_id as usize + i))
.collect();
RecordBatch::try_new(
Arc::new(schema.clone()),
vec![
Arc::new(Int64Array::from_iter_values(
start_id..start_id + num_rows as i64,
)),
Arc::new(vector_array),
Arc::new(StringArray::from_iter_values(texts)),
],
)
.unwrap()
}
#[tokio::test]
async fn test_initialize_mem_wal_records_writer_config_defaults() {
let vector_dim = 128;
let schema = create_test_schema(vector_dim);
let uri = format!("memory://test_writer_config_defaults_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let writer_config = ShardWriterConfig::default()
.with_durable_write(false)
.with_max_memtable_size(8 * 1024 * 1024);
dataset
.initialize_mem_wal()
.writer_config_defaults(writer_config)
.add_writer_config_default("custom_knob", "custom_value")
.execute()
.await
.expect("Failed to initialize MemWAL");
let details = dataset
.mem_wal_index_details()
.await
.expect("Failed to read MemWAL index details")
.expect("MemWAL index details should exist");
let defaults = &details.writer_config_defaults;
assert_eq!(
defaults.get("durable_write").map(String::as_str),
Some("false")
);
assert_eq!(
defaults.get("max_memtable_size").map(String::as_str),
Some("8388608")
);
assert_eq!(
defaults
.get("max_wal_flush_interval_ms")
.map(String::as_str),
Some("100")
);
assert!(defaults.contains_key("enable_memtable"));
assert_eq!(
defaults.get("custom_knob").map(String::as_str),
Some("custom_value")
);
assert!(!defaults.contains_key("shard_id"));
assert!(!defaults.contains_key("shard_spec_id"));
}
#[tokio::test]
async fn test_mem_wal_writer_with_multi_segment_index() {
use lance_index::optimize::OptimizeOptions;
let vector_dim = 32;
let schema = create_test_schema(vector_dim);
let uri = format!(
"shared-memory://multi-segment-index-{}/",
Uuid::new_v4().simple()
);
let initial = create_test_batch(&schema, 0, 256, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let vector_params = VectorIndexParams::ivf_flat(1, MetricType::L2);
dataset
.create_index(
&["vector"],
IndexType::Vector,
Some("vector_idx".to_string()),
&vector_params,
true,
)
.await
.expect("Failed to create vector index");
let appended = create_test_batch(&schema, 256, 256, vector_dim);
let append_batches = RecordBatchIterator::new([Ok(appended)], schema.clone());
dataset
.append(append_batches, None)
.await
.expect("Failed to append fragment");
dataset
.optimize_indices(&OptimizeOptions::append())
.await
.expect("Failed to append index delta");
assert_eq!(
dataset
.load_indices_by_name("vector_idx")
.await
.unwrap()
.len(),
2,
"expected two physical segments for the maintained index"
);
dataset
.initialize_mem_wal()
.maintained_indexes(["vector_idx"])
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let writer = dataset
.mem_wal_writer(
shard_id,
ShardWriterConfig::new(shard_id).with_durable_write(false),
)
.await
.expect("mem_wal_writer must accept a multi-segment maintained index");
writer
.put(vec![create_test_batch(&schema, 200, 10, vector_dim)])
.await
.unwrap();
assert_eq!(writer.memtable_stats().await.unwrap().row_count, 10);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_flush_all_tombstone_generation_skips_empty_hnsw() {
let vector_dim = 32;
let schema = create_test_schema(vector_dim);
let uri = format!("shared-memory://tombstone-{}/", Uuid::new_v4().simple());
let initial = create_test_batch(&schema, 0, 256, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.unwrap();
dataset
.create_index(
&["vector"],
IndexType::Vector,
Some("vector_idx".to_string()),
&VectorIndexParams::ivf_flat(1, MetricType::L2),
true,
)
.await
.unwrap();
dataset
.initialize_mem_wal()
.maintained_indexes(["vector_idx"])
.execute()
.await
.unwrap();
let shard_id = Uuid::new_v4();
let writer = dataset
.mem_wal_writer(
shard_id,
ShardWriterConfig::new(shard_id).with_durable_write(false),
)
.await
.unwrap();
let keys = RecordBatch::try_new(
Arc::new(ArrowSchema::new(vec![Field::new(
"id",
DataType::Int64,
false,
)])),
vec![Arc::new(Int64Array::from(vec![0_i64]))],
)
.unwrap();
writer.delete(vec![keys]).await.unwrap();
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.unwrap();
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
assert_eq!(
manifest.sstables.len(),
1,
"the all-tombstone generation must still flush"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_mem_wal_maintained_fts_v1_flush_preserves_format() {
use tempfile::TempDir;
let vector_dim = 32;
let schema = create_test_schema(vector_dim);
let temp_dir = TempDir::new().expect("Failed to create temp dir");
let uri = format!("file://{}", temp_dir.path().display());
let initial = create_test_batch(&schema, 0, 16, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let fts_params =
InvertedIndexParams::default().format_version(InvertedListFormatVersion::V1);
dataset
.create_index(
&["text"],
IndexType::Inverted,
Some("text_fts".to_string()),
&fts_params,
false,
)
.await
.expect("Failed to create v1 FTS index");
let base_indices = dataset.load_indices().await.unwrap();
assert_eq!(base_indices.len(), 1);
assert_eq!(base_indices[0].index_version, 1);
dataset
.initialize_mem_wal()
.maintained_indexes(["text_fts"])
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id).with_durable_write(true);
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("Failed to create MemWAL writer");
writer
.put(vec![create_test_batch(&schema, 1_000, 3, vector_dim)])
.await
.expect("Failed to write MemWAL batch");
writer.close().await.expect("Failed to close writer");
let (store, base_path) = lance_io::object_store::ObjectStore::from_uri(&uri)
.await
.expect("Failed to open store");
let manifest_store =
super::super::manifest::ShardManifestStore::new(store, &base_path, shard_id, 2);
let manifest = manifest_store
.latest()
.await
.expect("Failed to read manifest")
.expect("Manifest should exist");
assert_eq!(manifest.sstables.len(), 1);
let sstable = &manifest.sstables[0];
let gen_uri = format!("{}/_mem_wal/{}/{}", uri, shard_id, sstable.path);
let sstable = Dataset::open(&gen_uri)
.await
.expect("Failed to open SSTable");
let sstable_indices = sstable.load_indices().await.unwrap();
assert_eq!(sstable_indices.len(), 1);
assert_eq!(sstable_indices[0].name, "text_fts");
assert_eq!(
sstable_indices[0].index_version, 1,
"maintained v1 FTS index must flush as v1"
);
let results = sstable
.scan()
.full_text_search(FullTextSearchQuery::new("Sample".to_owned()))
.unwrap()
.try_into_batch()
.await
.unwrap();
assert_eq!(results.num_rows(), 3);
}
#[tokio::test]
async fn test_writer_hnsw_params_override() {
use lance_index::vector::hnsw::builder::HnswBuildParams;
let vector_dim = 32;
let schema = create_test_schema(vector_dim);
let uri = format!(
"shared-memory://writer-hnsw-params-{}/",
Uuid::new_v4().simple()
);
let initial = create_test_batch(&schema, 0, 256, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let vector_params = VectorIndexParams::ivf_flat(1, MetricType::L2);
dataset
.create_index(
&["vector"],
IndexType::Vector,
Some("vector_idx".to_string()),
&vector_params,
true,
)
.await
.expect("Failed to create vector index");
let configured = ShardWriterConfig::new(Uuid::new_v4()).with_hnsw_params(
"vector_idx",
HnswBuildParams::default().num_edges(7).ef_construction(48),
);
dataset
.initialize_mem_wal()
.maintained_indexes(["vector_idx"])
.writer_config_defaults(configured)
.execute()
.await
.expect("Failed to initialize MemWAL");
let defaults = dataset
.mem_wal_index_details()
.await
.unwrap()
.expect("MemWAL details should exist")
.writer_config_defaults;
assert_eq!(
defaults
.get("hnsw.vector_idx.num_edges")
.map(String::as_str),
Some("7")
);
assert_eq!(
defaults
.get("hnsw.vector_idx.ef_construction")
.map(String::as_str),
Some("48")
);
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id)
.with_durable_write(false)
.with_hnsw_params(
"vector_idx",
HnswBuildParams::default().num_edges(7).ef_construction(48),
);
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("mem_wal_writer must accept the HNSW-param override");
writer
.put(vec![create_test_batch(&schema, 300, 10, vector_dim)])
.await
.unwrap();
let active = writer.active_memtable_ref().await.unwrap();
let hnsw = active
.index_store
.get_hnsw("vector_idx")
.expect("maintained HNSW index should exist in the memtable");
assert_eq!(hnsw.build_params().m, 7);
assert_eq!(hnsw.build_params().ef_construction, 48);
drop(active);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_initialize_mem_wal_bucket_sharding() {
let vector_dim = 128;
let schema = create_test_schema(vector_dim);
let uri = format!("memory://test_bucket_sharding_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let result = dataset
.initialize_mem_wal()
.bucket_sharding("id", 0)
.execute()
.await;
assert!(result.is_err(), "num_buckets = 0 should be rejected");
dataset
.initialize_mem_wal()
.bucket_sharding("text", 8)
.execute()
.await
.expect("Failed to initialize MemWAL");
let details = dataset
.mem_wal_index_details()
.await
.expect("Failed to read MemWAL index details")
.expect("MemWAL index details should exist");
assert_eq!(details.num_shards, 8);
assert_eq!(details.sharding_specs.len(), 1);
let field = &details.sharding_specs[0].fields[0];
assert_eq!(field.transform.as_deref(), Some("bucket"));
assert_eq!(
field.parameters.get("num_buckets").map(String::as_str),
Some("8")
);
assert_eq!(field.source_ids.len(), 1);
let source_id = field.source_ids[0];
let source_field = dataset.schema().field("text").expect("text field exists");
assert_eq!(source_id, source_field.id);
}
#[tokio::test]
async fn test_initialize_mem_wal_bucket_sharding_without_primary_key() {
let vector_dim = 128;
let schema = create_append_only_schema(vector_dim);
let uri = format!(
"memory://test_bucket_sharding_no_primary_key_{}",
Uuid::new_v4()
);
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.initialize_mem_wal()
.bucket_sharding("id", 8)
.execute()
.await
.expect("Failed to initialize append-only MemWAL");
let details = dataset
.mem_wal_index_details()
.await
.expect("Failed to read MemWAL index details")
.expect("MemWAL index details should exist");
assert_eq!(details.num_shards, 8);
assert_eq!(details.sharding_specs.len(), 1);
let field = &details.sharding_specs[0].fields[0];
assert_eq!(field.transform.as_deref(), Some("bucket"));
assert_eq!(
field.parameters.get("num_buckets").map(String::as_str),
Some("8")
);
}
#[tokio::test]
async fn test_initialize_mem_wal_unsharded() {
let vector_dim = 128;
let schema = create_test_schema(vector_dim);
let uri = format!("memory://test_unsharded_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.initialize_mem_wal()
.unsharded()
.execute()
.await
.expect("Failed to initialize MemWAL");
let details = dataset
.mem_wal_index_details()
.await
.expect("Failed to read MemWAL index details")
.expect("MemWAL index details should exist");
assert_eq!(details.num_shards, 1);
assert_eq!(details.sharding_specs.len(), 1);
assert_eq!(
details.sharding_specs[0].fields[0].transform.as_deref(),
Some("unsharded")
);
}
#[tokio::test]
async fn test_initialize_mem_wal_identity_sharding() {
let vector_dim = 128;
let schema = create_test_schema(vector_dim);
let uri = format!("memory://test_identity_sharding_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let result = dataset
.initialize_mem_wal()
.identity_sharding("nonexistent")
.execute()
.await;
assert!(
result.is_err(),
"an unknown identity column should be rejected"
);
let result = dataset
.initialize_mem_wal()
.identity_sharding("vector")
.execute()
.await;
assert!(
result.is_err(),
"a non-scalar identity column should be rejected"
);
dataset
.initialize_mem_wal()
.identity_sharding("text")
.execute()
.await
.expect("Failed to initialize MemWAL");
let details = dataset
.mem_wal_index_details()
.await
.expect("Failed to read MemWAL index details")
.expect("MemWAL index details should exist");
assert_eq!(details.num_shards, 0);
assert_eq!(details.sharding_specs.len(), 1);
let field = &details.sharding_specs[0].fields[0];
assert_eq!(field.transform.as_deref(), Some("identity"));
assert_eq!(field.result_type.as_str(), "utf8");
assert_eq!(field.source_ids.len(), 1);
}
#[tokio::test]
async fn test_shard_writer_smoke() {
let vector_dim = 128;
let batch_size = 20;
let num_batches = 100;
let schema = create_test_schema(vector_dim);
let uri = format!("memory://test_shard_writer_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.initialize_mem_wal()
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id).with_durable_write(false);
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("Failed to create writer");
let batches: Vec<RecordBatch> = (0..num_batches)
.map(|i| create_test_batch(&schema, (i * batch_size) as i64, batch_size, vector_dim))
.collect();
writer.put(batches).await.expect("Failed to write");
writer.close().await.expect("Failed to close");
}
#[tokio::test]
async fn test_shard_writer_with_vector_index_searches_active_memtable() {
let vector_dim = 32;
let batch_size = 20;
let target_id = 1_000i64 + 37;
let schema = create_test_schema(vector_dim);
let uri = format!(
"shared-memory://shard-writer-hnsw-{}/",
Uuid::new_v4().simple()
);
let initial_batch = create_test_batch(&schema, 0, 256, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let vector_params = VectorIndexParams::ivf_flat(1, MetricType::L2);
dataset
.create_index(
&["vector"],
IndexType::Vector,
Some("vector_idx".to_string()),
&vector_params,
true,
)
.await
.expect("Failed to create base vector index");
dataset
.initialize_mem_wal()
.maintained_indexes(["vector_idx"])
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id).with_durable_write(true);
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("Failed to create writer");
let batches: Vec<RecordBatch> = (0..4)
.map(|i| {
create_test_batch(
&schema,
1_000 + (i * batch_size) as i64,
batch_size,
vector_dim,
)
})
.collect();
writer.put(batches).await.expect("Failed to write");
let query = Float32Array::from_iter_values(
(0..vector_dim as usize).map(|d| (target_id as f32 * 0.1 + d as f32 * 0.01).sin()),
);
let mut scanner = writer.scan().await.unwrap();
scanner.nearest("vector", &query, 80).unwrap();
let result = scanner.try_into_batch().await.expect("Failed to scan");
assert!(result.num_rows() > 0, "vector query returned no rows");
let id_col = result
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let dist_col = result
.column_by_name("_distance")
.unwrap()
.as_any()
.downcast_ref::<Float32Array>()
.unwrap();
let target_idx = (0..result.num_rows())
.find(|&idx| id_col.value(idx) == target_id)
.expect("target vector was not returned by active MemTable HNSW search");
assert!(
dist_col.value(target_idx) < 1e-6,
"expected self-match distance near zero, got {}",
dist_col.value(target_idx)
);
writer.close().await.expect("Failed to close");
}
#[tokio::test]
#[ignore]
async fn test_shard_writer_s3_ivfpq() {
let prefix = std::env::var("DATASET_PREFIX").expect("DATASET_PREFIX not set");
let vector_dim = 512;
let batch_size = 20;
let num_batches = 10000;
let num_partitions = 16;
let num_sub_vectors = 64;
let schema = create_test_schema(vector_dim);
let uri = format!(
"{}/test_s3_{}",
prefix.trim_end_matches('/'),
Uuid::new_v4()
);
let initial_batch = create_test_batch(&schema, 0, 1000, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let scalar_params = ScalarIndexParams::default();
dataset
.create_index(
&["id"],
IndexType::BTree,
Some("id_btree".to_string()),
&scalar_params,
false,
)
.await
.expect("Failed to create BTree index");
let fts_params = InvertedIndexParams::default();
dataset
.create_index(
&["text"],
IndexType::Inverted,
Some("text_fts".to_string()),
&fts_params,
false,
)
.await
.expect("Failed to create FTS index");
let ivf_params = IvfBuildParams {
num_partitions: Some(num_partitions),
..Default::default()
};
let pq_params = PQBuildParams {
num_sub_vectors,
num_bits: 8,
..Default::default()
};
let vector_params =
VectorIndexParams::with_ivf_pq_params(MetricType::L2, ivf_params, pq_params);
dataset
.create_index(
&["vector"],
IndexType::Vector,
Some("vector_idx".to_string()),
&vector_params,
true,
)
.await
.expect("Failed to create IVF-PQ index");
dataset
.initialize_mem_wal()
.maintained_indexes(["id_btree", "text_fts", "vector_idx"])
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id).with_durable_write(false);
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("Failed to create writer");
let batches: Vec<RecordBatch> = (0..num_batches)
.map(|i| create_test_batch(&schema, (i * batch_size) as i64, batch_size, vector_dim))
.collect();
writer.put(batches).await.expect("Failed to write");
writer.close().await.expect("Failed to close");
}
#[tokio::test]
async fn test_shard_writer_e2e_correctness() {
use std::time::Duration;
use tempfile::TempDir;
let vector_dim = 32;
let rows_per_batch = 50;
let num_write_rounds = 3;
let batches_per_round = 3;
let temp_dir = TempDir::new().expect("Failed to create temp dir");
let uri = format!("file://{}", temp_dir.path().display());
let schema = create_test_schema(vector_dim);
let initial_batch = create_test_batch(&schema, 0, 500, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.create_index(
&["id"],
IndexType::BTree,
Some("id_btree".to_string()),
&ScalarIndexParams::default(),
false,
)
.await
.expect("Failed to create BTree index");
dataset
.initialize_mem_wal()
.maintained_indexes(["id_btree"])
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let config = ShardWriterConfig::new(shard_id)
.with_durable_write(true) .with_max_memtable_size(50 * 1024) .with_max_wal_buffer_size(10 * 1024) .with_max_wal_flush_interval(Duration::from_millis(50));
let writer = dataset
.mem_wal_writer(shard_id, config)
.await
.expect("Failed to create writer");
let mut total_rows_written = 0i64;
for _round in 0..num_write_rounds {
let start_id = 500 + total_rows_written;
let batches_to_write: Vec<RecordBatch> = (0..batches_per_round)
.map(|i| {
create_test_batch(
&schema,
start_id + (i * rows_per_batch) as i64,
rows_per_batch,
vector_dim,
)
})
.collect();
writer.put(batches_to_write).await.expect("Failed to write");
total_rows_written += (batches_per_round * rows_per_batch) as i64;
tokio::time::sleep(Duration::from_millis(150)).await;
}
writer.close().await.expect("Failed to close");
let mem_wal_dir = temp_dir.path().join("_mem_wal").join(shard_id.to_string());
assert!(mem_wal_dir.exists(), "MemWAL directory should exist");
let wal_dir = mem_wal_dir.join("wal");
assert!(wal_dir.exists(), "WAL directory should exist");
let wal_files: Vec<_> = std::fs::read_dir(&wal_dir)
.expect("Failed to read WAL dir")
.filter_map(|e| e.ok())
.collect();
assert!(
!wal_files.is_empty(),
"WAL directory should contain at least one file"
);
let manifest_dir = mem_wal_dir.join("manifest");
assert!(manifest_dir.exists(), "Manifest directory should exist");
let manifest_files: Vec<_> = std::fs::read_dir(&manifest_dir)
.expect("Failed to read manifest dir")
.filter_map(|e| e.ok())
.collect();
assert!(
!manifest_files.is_empty(),
"Manifest directory should contain at least one file"
);
let (store, base_path) = lance_io::object_store::ObjectStore::from_uri(&uri)
.await
.expect("Failed to open store");
let manifest_store =
super::super::manifest::ShardManifestStore::new(store, &base_path, shard_id, 2);
let manifest = manifest_store
.latest()
.await
.expect("Failed to read manifest")
.expect("Manifest should exist");
assert!(
!manifest.sstables.is_empty(),
"Should have at least one SSTable"
);
for sstable in &manifest.sstables {
let gen_path = temp_dir
.path()
.join("_mem_wal")
.join(shard_id.to_string())
.join(&sstable.path);
assert!(
gen_path.exists(),
"SSTable directory should exist at {:?}",
gen_path
);
let gen_contents_count = std::fs::read_dir(&gen_path)
.expect("Failed to read gen dir")
.filter_map(|e| e.ok())
.count();
assert!(
gen_contents_count > 0,
"Generation directory should have files"
);
}
for wal_file in wal_files.iter().take(1) {
let wal_path = wal_file.path();
let file_name = wal_path.file_name().unwrap().to_string_lossy();
assert!(
file_name.ends_with(".arrow"),
"WAL file should have .arrow extension"
);
}
let dataset = Dataset::open(&uri).await.expect("Failed to reopen dataset");
let new_shard_id = Uuid::new_v4();
let new_config = ShardWriterConfig::new(new_shard_id).with_durable_write(true);
let new_writer = dataset
.mem_wal_writer(new_shard_id, new_config)
.await
.expect("Failed to create new writer");
let verify_batch = create_test_batch(&schema, 10000, 10, vector_dim);
new_writer
.put(vec![verify_batch])
.await
.expect("Failed to write to new shard");
let scanner = new_writer.scan().await.unwrap();
let result = scanner.try_into_batch().await.expect("Failed to scan");
assert_eq!(result.num_rows(), 10, "New shard should have 10 rows");
new_writer
.close()
.await
.expect("Failed to close new writer");
}
#[tokio::test]
async fn test_flush_and_read_with_path_bound_object_store() {
use crate::dataset::mem_wal::scanner::{LsmScanner, ShardSnapshot};
use futures::TryStreamExt;
use lance_io::object_store::ObjectStoreParams;
use tempfile::TempDir;
let vector_dim = 8;
let schema = create_test_schema(vector_dim);
let temp_dir = TempDir::new().unwrap();
let uri = format!("file://{}", temp_dir.path().display());
let initial = create_test_batch(&schema, 0, 16, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.initialize_mem_wal()
.execute()
.await
.expect("Failed to initialize MemWAL");
#[allow(deprecated)]
let store_params = ObjectStoreParams {
object_store: Some((
Arc::new(object_store::local::LocalFileSystem::new()),
url::Url::parse(&uri).unwrap(),
)),
..Default::default()
};
let dataset = dataset.with_object_store(dataset.object_store.clone(), Some(store_params));
let shard_id = Uuid::new_v4();
let writer = dataset
.mem_wal_writer(shard_id, ShardWriterConfig::new(shard_id))
.await
.expect("Failed to create writer");
writer
.put(vec![create_test_batch(&schema, 1_000, 8, vector_dim)])
.await
.expect("Failed to write");
writer.force_seal_active().await.unwrap();
writer
.wait_for_flush_drain()
.await
.expect("flush must not be redirected at the base table");
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
assert_eq!(manifest.sstables.len(), 1);
let sstable = manifest.sstables[0].clone();
let gen_uri = format!("{}/_mem_wal/{}/{}", uri, shard_id, sstable.path);
let generation = Dataset::open(&gen_uri)
.await
.expect("generation must exist at its own path");
assert_eq!(generation.count_rows(None).await.unwrap(), 8);
let base = Dataset::open(&uri).await.unwrap();
assert_eq!(
base.count_rows(None).await.unwrap(),
16,
"the generation write must not land in the base table"
);
let snapshot = ShardSnapshot::new(shard_id)
.with_current_generation(manifest.current_generation)
.with_sstable(sstable.generation, sstable.path.clone());
let scanner = LsmScanner::new(Arc::new(dataset), vec![snapshot], vec!["id".to_string()]);
let rows: usize = scanner
.try_into_stream()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.expect("scan must open the generation, not the base")
.iter()
.map(|batch| batch.num_rows())
.sum();
assert_eq!(rows, 24);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_store_params_reach_generation_write_and_read() {
use crate::dataset::builder::DatasetBuilder;
use crate::dataset::mem_wal::scanner::{LsmScanner, ShardSnapshot};
use crate::dataset::mem_wal::test_util::observable_store_params;
use futures::TryStreamExt;
use tempfile::TempDir;
let vector_dim = 8;
let schema = create_test_schema(vector_dim);
let temp_dir = TempDir::new().unwrap();
let uri = format!("file://{}", temp_dir.path().display());
let initial = create_test_batch(&schema, 0, 16, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
let (store_params, controls) = observable_store_params();
let mut dataset = DatasetBuilder::from_uri(&uri)
.with_store_params(store_params)
.load()
.await
.expect("Failed to open dataset");
dataset
.initialize_mem_wal()
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let writer = dataset
.mem_wal_writer(shard_id, ShardWriterConfig::new(shard_id))
.await
.expect("Failed to create writer");
writer
.put(vec![create_test_batch(&schema, 1_000, 8, vector_dim)])
.await
.expect("Failed to write");
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.expect("flush failed");
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
assert_eq!(manifest.sstables.len(), 1);
let sstable = manifest.sstables[0].clone();
let gen_manifest = format!("{}/_versions", sstable.path);
assert!(
controls.wrote_under(&gen_manifest),
"the flush must write the generation through the base's store params, \
not a store resolved from the generation URI alone"
);
let snapshot = ShardSnapshot::new(shard_id)
.with_current_generation(manifest.current_generation)
.with_sstable(sstable.generation, sstable.path.clone());
let scanner = LsmScanner::new(Arc::new(dataset), vec![snapshot], vec!["id".to_string()]);
let rows: usize = scanner
.try_into_stream()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.expect("scan failed")
.iter()
.map(|batch| batch.num_rows())
.sum();
assert_eq!(rows, 24);
assert!(
controls.read_under(&format!("{}/data/", sstable.path)),
"the scan must read the generation through the base's store params"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_fresh_tier_scan_with_path_bound_object_store() {
use crate::dataset::mem_wal::scanner::{LsmScanner, ShardSnapshot};
use futures::TryStreamExt;
use lance_io::object_store::ObjectStoreParams;
use tempfile::TempDir;
let vector_dim = 8;
let schema = create_test_schema(vector_dim);
let temp_dir = TempDir::new().unwrap();
let uri = format!("file://{}", temp_dir.path().display());
let initial = create_test_batch(&schema, 0, 16, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
dataset
.initialize_mem_wal()
.execute()
.await
.expect("Failed to initialize MemWAL");
let shard_id = Uuid::new_v4();
let writer = dataset
.mem_wal_writer(shard_id, ShardWriterConfig::new(shard_id))
.await
.expect("Failed to create writer");
writer
.put(vec![create_test_batch(&schema, 1_000, 8, vector_dim)])
.await
.expect("Failed to write");
writer.force_seal_active().await.unwrap();
writer.wait_for_flush_drain().await.expect("flush failed");
let manifest = writer.manifest().await.unwrap().expect("manifest exists");
let sstable = manifest.sstables[0].clone();
let snapshot = ShardSnapshot::new(shard_id)
.with_current_generation(manifest.current_generation)
.with_sstable(sstable.generation, sstable.path.clone());
#[allow(deprecated)]
let store_params = ObjectStoreParams {
object_store: Some((
Arc::new(object_store::local::LocalFileSystem::new()),
url::Url::parse(&uri).unwrap(),
)),
..Default::default()
};
let arrow_schema: Arc<ArrowSchema> = schema.clone();
let batches = LsmScanner::without_base_table(
arrow_schema,
uri.clone(),
vec![snapshot],
vec!["id".to_string()],
)
.with_session(dataset.session())
.with_store_params(store_params)
.try_into_stream()
.await
.unwrap()
.try_collect::<Vec<_>>()
.await
.expect("scan must open the generation, not the base");
let rows: usize = batches.iter().map(|batch| batch.num_rows()).sum();
assert_eq!(
rows, 8,
"fresh tier holds only the 8 WAL rows; 16 means the generation open \
was redirected at the base table"
);
let ids: Vec<i64> = batches
.iter()
.flat_map(|batch| {
batch
.column_by_name("id")
.unwrap()
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.values()
.to_vec()
})
.collect();
assert!(
ids.iter().all(|id| (1_000..1_008).contains(id)),
"expected the WAL's own rows, got {ids:?}"
);
writer.close().await.unwrap();
}
#[tokio::test]
async fn test_initialize_mem_wal_rejects_a_nullable_primary_key() {
let vector_dim = 128;
let schema = create_append_only_schema(vector_dim);
let uri = format!("memory://test_mem_wal_nullable_pk_{}", Uuid::new_v4());
let initial_batch = create_test_batch(&schema, 0, 100, vector_dim);
let batches = RecordBatchIterator::new([Ok(initial_batch)], schema.clone());
let mut dataset = Dataset::write(batches, &uri, Some(WriteParams::default()))
.await
.expect("Failed to create dataset");
{
let manifest = Arc::make_mut(&mut dataset.manifest);
let id_field = manifest
.schema
.fields
.iter_mut()
.find(|field| field.name == "id")
.expect("schema has an id column");
id_field.unenforced_primary_key_position = Some(1);
id_field.nullable = true;
}
let err = dataset
.initialize_mem_wal()
.unsharded()
.execute()
.await
.expect_err("MemWAL must not enable on a nullable primary key");
assert!(
err.to_string().contains("must not be nullable"),
"unexpected error: {err}"
);
}
}