#[cfg(feature = "write-support")]
use crate::error::{Error, Result};
#[cfg(feature = "write-support")]
use crate::schema::TableSchema;
#[cfg(feature = "write-support")]
use crate::storage::write_engine::mutation::{
ClusteringKey, DecoratedKey, PartitionTombstone, RangeTombstone,
};
#[cfg(feature = "write-support")]
use crate::storage::write_engine::reconcile_rules;
#[cfg(feature = "write-support")]
use crate::types::Value;
#[cfg(feature = "write-support")]
use std::cmp::{Ordering, Reverse};
#[cfg(feature = "write-support")]
use std::collections::{BinaryHeap, VecDeque};
#[cfg(feature = "write-support")]
use std::path::{Path, PathBuf};
#[cfg(feature = "write-support")]
use std::time::{Duration, Instant};
mod model;
#[cfg(feature = "write-support")]
pub use model::{CellData, ComplexDeletion, MergeEntry, MergeStats, MergeStep, RowData};
mod read_assembly;
#[cfg(feature = "write-support")]
pub use read_assembly::assemble_read_cells;
#[cfg(feature = "write-support")]
mod point_read;
#[cfg(feature = "write-support")]
pub use point_read::{build_single_partition_merger, build_single_partition_merger_from_readers};
#[cfg(feature = "write-support")]
mod from_readers;
#[cfg(feature = "write-support")]
mod fully_expired;
#[cfg(feature = "write-support")]
pub use fully_expired::fully_expired_sstables;
#[cfg(feature = "write-support")]
pub(crate) use fully_expired::{reclaim_dropped_whole, split_merge_and_dropped};
#[cfg(feature = "write-support")]
pub mod repair_state;
#[cfg(feature = "write-support")]
pub use repair_state::{classify_inputs, RepairState};
#[cfg(feature = "write-support")]
mod carriers;
#[cfg(feature = "write-support")]
mod reconcile;
#[cfg(feature = "write-support")]
mod streaming;
#[cfg(feature = "write-support")]
pub(crate) use streaming::PartitionReconcileCheckpoint;
#[cfg(feature = "write-support")]
pub use streaming::{StreamingMerger, StreamingStep};
#[cfg(feature = "write-support")]
mod schema_order;
#[cfg(feature = "write-support")]
impl reconcile_rules::ReconcileCell for CellData {
fn timestamp(&self) -> i64 {
self.timestamp
}
fn is_tombstone(&self) -> bool {
KWayMerger::is_cell_tombstone(self)
}
}
#[cfg(feature = "write-support")]
#[derive(Clone)]
enum RangeCut {
Bottom,
At {
key: ClusteringKey,
after: bool,
},
Top,
}
#[cfg(feature = "write-support")]
struct RunReader {
reader: Box<dyn SSTableRowIterator>,
buffer: VecDeque<MergeEntry>,
buffer_size: usize,
exhausted: bool,
}
#[cfg(feature = "write-support")]
impl std::fmt::Debug for RunReader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RunReader")
.field("buffer_len", &self.buffer.len())
.field("buffer_size", &self.buffer_size)
.field("exhausted", &self.exhausted)
.finish()
}
}
#[cfg(feature = "write-support")]
impl RunReader {
const DEFAULT_BUFFER_SIZE: usize = 8 * 1024;
fn new(reader: Box<dyn SSTableRowIterator>) -> Self {
Self {
reader,
buffer: VecDeque::new(),
buffer_size: Self::DEFAULT_BUFFER_SIZE,
exhausted: false,
}
}
#[cfg(test)]
fn peek(&mut self) -> Result<Option<&MergeEntry>> {
if self.buffer.is_empty() && !self.exhausted {
self.refill_buffer()?;
}
Ok(self.buffer.front())
}
fn advance(&mut self) -> Result<Option<MergeEntry>> {
if let Some(entry) = self.buffer.pop_front() {
return Ok(Some(entry));
}
if !self.exhausted {
self.refill_buffer()?;
Ok(self.buffer.pop_front())
} else {
Ok(None)
}
}
fn is_exhausted(&self) -> bool {
self.exhausted && self.buffer.is_empty()
}
fn refill_buffer(&mut self) -> Result<()> {
let mut bytes_buffered = 0;
while bytes_buffered < self.buffer_size {
match self.reader.next() {
Some(Ok(entry)) => {
bytes_buffered += Self::estimate_entry_size(&entry);
self.buffer.push_back(entry);
}
Some(Err(e)) => return Err(e),
None => {
self.exhausted = true;
break;
}
}
}
Ok(())
}
fn estimate_entry_size(entry: &MergeEntry) -> usize {
let base_size = std::mem::size_of::<MergeEntry>();
let key_size = entry.key.key.len();
let clustering_size = entry
.clustering_key
.as_ref()
.map(|ck| {
ck.columns
.iter()
.map(|(name, value)| name.len() + Self::estimate_value_size(value))
.sum()
})
.unwrap_or(0);
let data_size = match &entry.row_data {
RowData::Live { cells } => cells
.iter()
.map(|cell| {
std::mem::size_of::<CellData>()
+ cell.column.len()
+ Self::estimate_value_size(&cell.value)
+ cell.cell_path.as_ref().map_or(0, |p| p.len())
})
.sum(),
RowData::Tombstone { .. } => 16,
};
let complex_deletion_size: usize = entry
.complex_deletions
.iter()
.map(|cd| std::mem::size_of::<ComplexDeletion>() + cd.column.len())
.sum();
base_size + key_size + clustering_size + data_size + complex_deletion_size
}
fn estimate_value_size(value: &Value) -> usize {
match value {
Value::Null => 0,
Value::Boolean(_) => 1,
Value::TinyInt(_) => 1,
Value::SmallInt(_) => 2,
Value::Integer(_) => 4,
Value::BigInt(_) | Value::Counter(_) | Value::Timestamp(_) | Value::Time(_) => 8,
Value::Float32(_) => 4,
Value::Float(_) => 8,
Value::Text(s) => s.len() + std::mem::size_of::<String>(),
Value::Blob(b) => b.len() + std::mem::size_of::<Vec<u8>>(),
Value::Uuid(_) => 16,
Value::Inet(b) => b.len() + std::mem::size_of::<Vec<u8>>(),
Value::Varint(b) => b.len() + std::mem::size_of::<Vec<u8>>(),
Value::Decimal { unscaled, .. } => unscaled.len() + 4 + std::mem::size_of::<Vec<u8>>(),
Value::Date(_) => 4,
Value::Duration { .. } => 20,
_ => 32, }
}
}
#[cfg(feature = "write-support")]
pub trait SSTableRowIterator: Send {
fn next(&mut self) -> Option<Result<MergeEntry>>;
}
mod async_bridge;
#[cfg(feature = "write-support")]
pub(crate) use async_bridge::block_on_async;
#[cfg(feature = "write-support")]
struct SSTableRowIteratorAdapter {
receiver:
Option<std::sync::mpsc::Receiver<std::result::Result<MergeEntry, MergeProducerError>>>,
producer: Option<std::thread::JoinHandle<()>>,
scan_cancel: crate::storage::scan_cancel::ScanCancel,
sent_count: std::sync::Arc<std::sync::atomic::AtomicI64>,
received_count: i64,
}
#[cfg(feature = "write-support")]
const RECV_CANCEL_POLL: std::time::Duration = std::time::Duration::from_millis(50);
#[cfg(feature = "write-support")]
#[derive(Debug)]
enum MergeProducerError {
Cancelled,
Other(String),
}
#[cfg(feature = "write-support")]
impl From<Error> for MergeProducerError {
fn from(e: Error) -> Self {
match e {
Error::Cancelled => MergeProducerError::Cancelled,
other => MergeProducerError::Other(other.to_string()),
}
}
}
#[cfg(feature = "write-support")]
const STREAMING_CHANNEL_CAPACITY: usize = 256;
#[cfg(feature = "write-support")]
mod producer_gauge;
#[cfg(feature = "write-support")]
mod channel_depth;
#[cfg(all(test, feature = "write-support"))]
mod teardown_tests;
#[cfg(all(test, feature = "write-support"))]
mod clone_regression_tests;
#[cfg(all(test, feature = "write-support"))]
mod reconcile_microalloc_tests;
#[cfg(all(test, feature = "write-support"))]
mod range_shadowing_binsearch_tests;
#[cfg(feature = "write-support")]
impl SSTableRowIteratorAdapter {
fn open(
path: &Path,
run_index: usize,
schema: &TableSchema,
udt_registry: Option<crate::schema::UdtRegistry>,
scan_cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
let path_buf = path.to_path_buf();
let schema = schema.clone();
let adapter_cancel = scan_cancel.clone();
let (sender, receiver) = std::sync::mpsc::sync_channel(STREAMING_CHANNEL_CAPACITY);
let sent_count = std::sync::Arc::new(std::sync::atomic::AtomicI64::new(0));
let producer_sent_count = sent_count.clone();
producer_gauge::spawned();
let producer = match std::thread::Builder::new().spawn(move || {
Self::producer_thread(
path_buf,
run_index,
schema,
udt_registry,
scan_cancel,
sender,
producer_sent_count,
);
}) {
Ok(handle) => handle,
Err(e) => {
producer_gauge::rollback();
return Err(Error::Storage(format!(
"streaming producer: failed to spawn thread: {}",
e
)));
}
};
Ok(Self {
receiver: Some(receiver),
producer: Some(producer),
scan_cancel: adapter_cancel,
sent_count,
received_count: 0,
})
}
fn producer_thread(
path_buf: PathBuf,
run_index: usize,
schema: TableSchema,
udt_registry: Option<crate::schema::UdtRegistry>,
scan_cancel: crate::storage::scan_cancel::ScanCancel,
sender: std::sync::mpsc::SyncSender<std::result::Result<MergeEntry, MergeProducerError>>,
sent_count: std::sync::Arc<std::sync::atomic::AtomicI64>,
) {
let _thread_guard = producer_gauge::ProducerThreadGuard;
let error_sender = sender.clone();
let stream_result = (|| -> Result<()> {
use crate::config::DiskAccessMode;
use crate::platform::Platform;
use crate::Config;
use std::sync::Arc;
let mut config = Config::default();
config.storage.use_mmap = false;
config.storage.disk_access_mode = DiskAccessMode::Buffered;
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| {
Error::Storage(format!(
"streaming producer: failed to create runtime: {}",
e
))
})?;
rt.block_on(async move {
let platform = Arc::new(Platform::new(&config).await?);
let mut reader = crate::storage::sstable::reader::SSTableReader::open(
&path_buf, &config, platform,
)
.await?;
if let Some(registry) = udt_registry {
reader.set_udt_registry(registry);
}
from_readers::drive_compaction_stream(
&reader,
run_index,
&schema,
&scan_cancel,
&sender,
sent_count.as_ref(),
)
.await
})
})();
if let Err(e) = stream_result {
let _ = error_sender.send(Err(MergeProducerError::from(e)));
}
}
fn build_merge_entry(
run_index: usize,
compaction_row: crate::storage::sstable::reader::compaction_row::CompactionRow,
schema: &TableSchema,
) -> Result<MergeEntry> {
use crate::storage::sstable::reader::compaction_row::CompactionRowData;
let crate::storage::sstable::reader::compaction_row::CompactionRow {
key,
row_timestamp,
row_data,
} = compaction_row;
let decorated_key = DecoratedKey::from_key_bytes(key.as_bytes().to_vec())?;
if let CompactionRowData::PartitionDelete {
deletion_time,
local_deletion_time,
} = row_data
{
return Ok(MergeEntry::new(
run_index,
decorated_key,
None,
deletion_time,
RowData::Tombstone {
deletion_time,
local_deletion_time,
},
)
.with_partition_deletion((deletion_time, local_deletion_time)));
}
if let CompactionRowData::RangeMarker {
start,
end,
deletion_time,
local_deletion_time,
} = row_data
{
let range = RangeTombstone {
start: Self::compaction_bound_to_mutation(start),
end: Self::compaction_bound_to_mutation(end),
deletion_time,
local_deletion_time,
};
return Ok(MergeEntry::new(
run_index,
decorated_key,
None,
deletion_time,
RowData::Live { cells: Vec::new() },
)
.with_range_deletion(range));
}
let clustering_key = Self::extract_clustering_key_from_compaction(&row_data, schema);
let (row_data, complex_deletions, row_deletion) =
Self::compaction_row_data_to_row_data(row_data, row_timestamp);
let entry = MergeEntry::new(
run_index,
decorated_key,
clustering_key,
row_timestamp,
row_data,
);
let entry = if complex_deletions.is_empty() {
entry
} else {
entry.with_complex_deletions(complex_deletions)
};
let entry = match row_deletion {
Some((deletion_time, ldt)) => entry.with_row_deletion(deletion_time, ldt),
None => entry,
};
Ok(entry)
}
fn compaction_bound_to_mutation(
bound: crate::storage::sstable::reader::compaction_row::CompactionBound,
) -> crate::storage::write_engine::mutation::ClusteringBound {
use crate::storage::sstable::reader::compaction_row::CompactionBound;
use crate::storage::write_engine::mutation::ClusteringBound;
match bound {
CompactionBound::Inclusive(cols) => {
ClusteringBound::Inclusive(ClusteringKey { columns: cols })
}
CompactionBound::Exclusive(cols) => {
ClusteringBound::Exclusive(ClusteringKey { columns: cols })
}
CompactionBound::Bottom => ClusteringBound::Bottom,
CompactionBound::Top => ClusteringBound::Top,
}
}
fn extract_clustering_key_from_compaction(
row_data: &crate::storage::sstable::reader::compaction_row::CompactionRowData,
schema: &TableSchema,
) -> Option<ClusteringKey> {
use crate::storage::sstable::reader::compaction_row::CompactionRowData;
if schema.clustering_keys.is_empty() {
return None;
}
match row_data {
CompactionRowData::Live { simple, .. } => {
let mut ck_columns: Vec<(String, Value)> =
Vec::with_capacity(schema.clustering_keys.len());
for ck_col in &schema.clustering_keys {
match simple
.iter()
.find(|c| c.column == ck_col.name)
.map(|c| (ck_col.name.clone(), c.value.clone()))
{
Some(pair) => ck_columns.push(pair),
None => return None,
}
}
Some(ClusteringKey {
columns: ck_columns,
})
}
CompactionRowData::Tombstone { clustering, .. } => {
if clustering.is_empty() {
return None;
}
let mut ck_columns: Vec<(String, Value)> =
Vec::with_capacity(schema.clustering_keys.len());
for ck_col in &schema.clustering_keys {
match clustering
.iter()
.find(|(name, _)| name == &ck_col.name)
.map(|(name, v)| (name.clone(), v.clone()))
{
Some(pair) => ck_columns.push(pair),
None => return None,
}
}
Some(ClusteringKey {
columns: ck_columns,
})
}
CompactionRowData::RangeMarker { .. } => None,
CompactionRowData::PartitionDelete { .. } => None,
}
}
fn compaction_row_data_to_row_data(
row_data: crate::storage::sstable::reader::compaction_row::CompactionRowData,
_row_timestamp: i64,
) -> (RowData, Vec<ComplexDeletion>, Option<(i64, i32)>) {
use crate::storage::sstable::reader::compaction_row::CompactionRowData;
match row_data {
CompactionRowData::Tombstone {
deletion_time,
local_deletion_time,
clustering: _,
} => (
RowData::Tombstone {
deletion_time,
local_deletion_time,
},
Vec::new(),
None,
),
CompactionRowData::Live {
simple,
complex,
row_deletion,
} => {
let element_count: usize = complex.iter().map(|c| c.elements.len()).sum();
let mut cells = Vec::with_capacity(simple.len() + element_count);
let mut complex_deletions = Vec::new();
for sc in simple {
cells.push(CellData {
column: sc.column,
value: sc.value,
timestamp: sc.timestamp,
ttl: sc.ttl,
cell_path: None,
local_deletion_time: sc.local_deletion_time,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
});
}
for col in complex {
if let Some((marked_for_delete_at, ldt)) = col.complex_deletion {
complex_deletions.push(ComplexDeletion {
column: col.column.clone(),
marked_for_delete_at,
local_deletion_time: ldt,
});
}
for elem in col.elements {
let value = if elem.is_deleted {
Value::Tombstone(Box::new(crate::types::TombstoneInfo {
deletion_time: elem.timestamp,
tombstone_type: crate::types::TombstoneType::CellTombstone,
local_deletion_time: elem.local_deletion_time.unwrap_or(0) as i64,
ttl: None,
range_start: None,
range_end: None,
}))
} else {
elem.value.unwrap_or(Value::Null)
};
cells.push(CellData {
column: col.column.clone(),
value,
timestamp: elem.timestamp,
ttl: elem.ttl,
cell_path: Some(elem.cell_path),
local_deletion_time: elem.local_deletion_time,
is_complex_element: true,
is_deleted: elem.is_deleted,
has_empty_value: elem.has_empty_value,
});
}
}
(RowData::Live { cells }, complex_deletions, row_deletion)
}
CompactionRowData::RangeMarker { .. } => {
(RowData::Live { cells: Vec::new() }, Vec::new(), None)
}
CompactionRowData::PartitionDelete { .. } => {
(RowData::Live { cells: Vec::new() }, Vec::new(), None)
}
}
}
#[cfg(test)]
fn value_to_row_data(value: &crate::types::Value, row_timestamp: i64) -> Result<RowData> {
match value {
crate::types::Value::Tombstone(info) => Ok(RowData::Tombstone {
deletion_time: info.deletion_time,
local_deletion_time: info.local_deletion_time as i32,
}),
crate::types::Value::Map(map_entries) => {
let mut cells = Vec::with_capacity(map_entries.len());
for (key, val) in map_entries {
let column = match key {
crate::types::Value::Text(s) => String::from_utf8_lossy(s).into_owned(),
other => format!("{:?}", other),
};
let cell_ts = match val {
crate::types::Value::Tombstone(info) => info.deletion_time,
_ => row_timestamp,
};
cells.push(CellData {
column,
value: val.clone(),
timestamp: cell_ts,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
});
}
Ok(RowData::Live { cells })
}
other => Ok(RowData::Live {
cells: vec![CellData {
column: "value".to_string(),
value: other.clone(),
timestamp: row_timestamp,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
}),
}
}
}
#[cfg(feature = "write-support")]
impl SSTableRowIterator for SSTableRowIteratorAdapter {
fn next(&mut self) -> Option<Result<MergeEntry>> {
use std::sync::mpsc::RecvTimeoutError;
let receiver = self.receiver.as_ref()?;
loop {
if self.scan_cancel.is_cancelled() {
return Some(Err(Error::Cancelled));
}
match receiver.recv_timeout(RECV_CANCEL_POLL) {
Ok(Ok(entry)) => {
channel_depth::received();
self.received_count += 1;
crate::storage::sstable::work_counters::add_merge_run_entry_decoded();
return Some(Ok(entry));
}
Ok(Err(MergeProducerError::Cancelled)) => return Some(Err(Error::Cancelled)),
Ok(Err(MergeProducerError::Other(msg))) => {
return Some(Err(Error::Storage(format!(
"streaming merge producer error: {}",
msg
))))
}
Err(RecvTimeoutError::Timeout) => continue,
Err(RecvTimeoutError::Disconnected) => return None,
}
}
}
}
#[cfg(feature = "write-support")]
impl Drop for SSTableRowIteratorAdapter {
fn drop(&mut self) {
self.scan_cancel.cancel();
drop(self.receiver.take());
if let Some(handle) = self.producer.take() {
let _ = handle.join();
}
let residual =
self.sent_count.load(std::sync::atomic::Ordering::SeqCst) - self.received_count;
channel_depth::reconcile_residual(residual);
}
}
#[cfg(feature = "write-support")]
#[derive(Debug)]
pub struct KWayMerger {
runs: Vec<RunReader>,
heap: BinaryHeap<Reverse<schema_order::SchemaOrderedEntry>>,
current_partition: Option<DecoratedKey>,
schema: TableSchema,
schema_arc: std::sync::Arc<TableSchema>,
#[allow(dead_code)]
gc_before_secs: Option<i64>,
#[allow(dead_code)]
now_secs: Option<i64>,
purge_safe: bool,
max_purgeable_timestamp: Option<i64>,
}
#[cfg(feature = "write-support")]
#[derive(Debug)]
pub struct CompactReport {
pub output: crate::storage::sstable::writer::SSTableInfo,
pub stats: MergeStats,
}
#[cfg(feature = "write-support")]
fn stats_path_for(data_path: &Path) -> PathBuf {
let filename = data_path.file_name().and_then(|n| n.to_str()).unwrap_or("");
let stats_filename = filename.replace("Data.db", "Statistics.db");
data_path.parent().unwrap_or(data_path).join(stats_filename)
}
#[cfg(feature = "write-support")]
pub fn compute_baseline_min(input_paths: &[PathBuf]) -> (i64, i32, i32) {
let mut baseline_min_ts = i64::MAX;
let mut baseline_min_ldt = i32::MAX;
let mut baseline_min_ttl = i32::MAX;
for data_path in input_paths {
let stats_path = stats_path_for(data_path);
if !stats_path.exists() {
continue;
}
let stats_bytes = match std::fs::read(&stats_path) {
Ok(b) => b,
Err(e) => {
tracing::warn!(
"Could not read Statistics.db {:?} for baseline pre-seeding: {}",
stats_path,
e
);
continue;
}
};
match crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
) {
Ok((_, sstable_stats)) => {
let ts_stats = &sstable_stats.timestamp_stats;
baseline_min_ts = baseline_min_ts.min(ts_stats.min_timestamp);
let ldt_bits = ts_stats.min_deletion_time as u32 as i32;
let has_tombstone =
match crate::parser::repair_metadata::parse_stats_extras(&stats_bytes, None) {
Ok(extras) => !extras.tombstone_drop_times.is_empty(),
Err(e) => {
tracing::debug!(
"STATS-extras decode failed for {:?} during baseline seeding; \
conservatively including its LDT baseline: {:?}",
stats_path,
e
);
true
}
};
if has_tombstone {
baseline_min_ldt = baseline_min_ldt.min(ldt_bits);
}
if let Some(min_ttl) = ts_stats.min_ttl {
if min_ttl > 0 && min_ttl < i32::MAX as i64 {
baseline_min_ttl = baseline_min_ttl.min(min_ttl as i32);
}
}
}
Err(e) => {
tracing::warn!(
"Could not parse Statistics.db {:?} for baseline pre-seeding: {:?}",
stats_path,
e
);
}
}
}
(baseline_min_ts, baseline_min_ldt, baseline_min_ttl)
}
#[cfg(feature = "write-support")]
pub fn all_input_stats_readable(input_paths: &[PathBuf]) -> bool {
for data_path in input_paths {
let stats_path = stats_path_for(data_path);
if !stats_path.exists() {
return false;
}
let stats_bytes = match std::fs::read(&stats_path) {
Ok(b) => b,
Err(_) => return false,
};
if crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
)
.is_err()
{
return false;
}
}
true
}
pub(crate) fn compute_expiry_ttl_ldt_floor(input_paths: &[PathBuf]) -> Option<i32> {
let mut floor: Option<i32> = None;
for data_path in input_paths {
let stats_path = stats_path_for(data_path);
if !stats_path.exists() {
continue;
}
let stats_bytes = match std::fs::read(&stats_path) {
Ok(b) => b,
Err(e) => {
tracing::warn!(
"Could not read Statistics.db {:?} for expiry-TTL LDT floor: {}",
stats_path,
e
);
continue;
}
};
if let Ok((_, sstable_stats)) =
crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
)
{
let ts_stats = &sstable_stats.timestamp_stats;
let Some(max_ttl) = ts_stats.max_ttl.filter(|&t| t > 0) else {
continue;
};
let min_ldt_unsigned = i64::from(ts_stats.min_deletion_time as u32);
let input_floor_i64 = (min_ldt_unsigned - max_ttl).max(0);
let input_floor = input_floor_i64 as i32;
floor = Some(match floor {
Some(cur) => cur.min(input_floor),
None => input_floor,
});
}
}
floor
}
#[cfg(feature = "write-support")]
pub fn compute_max_purgeable_timestamp(outside_paths: &[PathBuf]) -> Option<i64> {
if outside_paths.is_empty() {
return None;
}
let mut min_ts = i64::MAX;
for data_path in outside_paths {
let stats_path = stats_path_for(data_path);
if !stats_path.exists() {
return None;
}
let stats_bytes = match std::fs::read(&stats_path) {
Ok(b) => b,
Err(e) => {
tracing::warn!(
"Could not read Statistics.db {:?} for max-purgeable bound: {}",
stats_path,
e
);
return None;
}
};
match crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
) {
Ok((_, sstable_stats)) => {
min_ts = min_ts.min(sstable_stats.timestamp_stats.min_timestamp);
}
Err(e) => {
tracing::warn!(
"Could not parse Statistics.db {:?} for max-purgeable bound: {:?}",
stats_path,
e
);
return None;
}
}
}
Some(min_ts)
}
#[cfg(feature = "write-support")]
#[derive(Debug, Default, Clone)]
pub struct UdtNormalizationPlan {
pub eligible_marshals: std::collections::HashMap<String, String>,
pub conflicts: std::collections::HashSet<String>,
pub observed: std::collections::HashSet<String>,
pub headers_verified: bool,
}
pub fn udt_columns_eligible_for_normalization(input_paths: &[PathBuf]) -> UdtNormalizationPlan {
use std::collections::{HashMap, HashSet};
const USERTYPE_MARKER: &str = "org.apache.cassandra.db.marshal.usertype(";
let mut usertype_marshals: HashMap<String, HashSet<String>> = HashMap::new();
let mut vetoed: HashSet<String> = HashSet::new();
for data_path in input_paths {
let stats_path = stats_path_for(data_path);
let Ok(stats_bytes) = std::fs::read(&stats_path) else {
return UdtNormalizationPlan::default();
};
match crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
) {
Ok((_, sstable_stats)) => {
for c in &sstable_stats.serialization_header_columns {
if c.column_type.to_lowercase().contains(USERTYPE_MARKER) {
usertype_marshals
.entry(c.name.clone())
.or_default()
.insert(c.column_type.clone());
} else {
vetoed.insert(c.name.clone());
}
}
}
Err(_) => return UdtNormalizationPlan::default(),
}
}
let mut plan = UdtNormalizationPlan {
headers_verified: true,
..UdtNormalizationPlan::default()
};
plan.observed.extend(vetoed.iter().cloned());
plan.observed.extend(usertype_marshals.keys().cloned());
for (column, marshals) in usertype_marshals {
if vetoed.contains(&column) || marshals.len() != 1 {
plan.conflicts.insert(column);
} else if let Some(marshal) = marshals.into_iter().next() {
plan.eligible_marshals.insert(column, marshal);
}
}
plan
}
pub(crate) fn apply_udt_marshals_from_inputs(
schema: &mut TableSchema,
input_paths: &[PathBuf],
registry: Option<&crate::schema::UdtRegistry>,
) -> Result<()> {
let plan = udt_columns_eligible_for_normalization(input_paths);
if !plan.conflicts.is_empty() {
let mut cols: Vec<&String> = plan.conflicts.iter().collect();
cols.sort();
return Err(Error::InvalidInput(format!(
"compaction inputs disagree on the encoding of UDT column(s) {cols:?}: some store \
complex UserType cells and others a simple cell (or a different UDT definition). \
Refusing to compact to avoid data loss; rewrite the divergent SSTable(s) first."
)));
}
let keyspace = schema.keyspace.clone();
for column in &mut schema.columns {
if let Some(marshal) = plan.eligible_marshals.get(&column.name) {
column.data_type = marshal.clone();
} else if plan.headers_verified && !plan.observed.contains(&column.name) {
if let Some(reg) = registry {
if let Some(marshal) =
crate::storage::sstable::writer::data_writer::resolve_bare_udt_marshal(
&column.data_type,
&keyspace,
reg,
)
{
column.data_type = marshal;
}
}
}
}
Ok(())
}
pub fn effective_compaction_schema(schema: &TableSchema, input_paths: &[PathBuf]) -> TableSchema {
use std::collections::HashSet;
let mut known: HashSet<String> = schema.columns.iter().map(|c| c.name.clone()).collect();
for k in &schema.partition_keys {
known.insert(k.name.clone());
}
for k in &schema.clustering_keys {
known.insert(k.name.clone());
}
let mut added: Vec<(String, String)> = Vec::new();
let mut seen: HashSet<String> = HashSet::new();
for data_path in input_paths {
let stats_path = stats_path_for(data_path);
if !stats_path.exists() {
continue;
}
let stats_bytes = match std::fs::read(&stats_path) {
Ok(b) => b,
Err(e) => {
tracing::warn!(
"effective_compaction_schema: cannot read Statistics.db {:?}: {}",
stats_path,
e
);
continue;
}
};
match crate::parser::enhanced_statistics_parser::parse_statistics_with_fallback(
&stats_bytes,
None,
) {
Ok((_, sstable_stats)) => {
for col in &sstable_stats.serialization_header_columns {
if col.is_static && !known.contains(&col.name) && seen.insert(col.name.clone())
{
added.push((col.name.clone(), col.column_type.clone()));
}
}
}
Err(e) => {
tracing::warn!(
"effective_compaction_schema: cannot parse Statistics.db {:?}: {:?}",
stats_path,
e
);
}
}
}
if added.is_empty() {
return schema.clone();
}
added.sort_by(|a, b| a.0.cmp(&b.0));
tracing::info!(
"effective_compaction_schema: re-adding {} static column(s) dropped from schema \
{}.{} but still present in input SSTable headers: {:?}",
added.len(),
schema.keyspace,
schema.table,
added.iter().map(|(n, _)| n.as_str()).collect::<Vec<_>>()
);
let mut effective = schema.clone();
for (name, data_type) in added {
effective.columns.push(crate::schema::Column {
name,
data_type,
nullable: true,
default: None,
is_static: true,
});
}
effective
}
#[cfg(feature = "write-support")]
pub(crate) fn compute_gc_before(schema: &TableSchema, now_secs: i64) -> Option<i64> {
const DEFAULT_GC_GRACE_SECONDS: i64 = 864_000;
match schema.comments.get("gc_grace_seconds") {
None => Some(now_secs - DEFAULT_GC_GRACE_SECONDS),
Some(s) => match s.trim().parse::<i64>() {
Ok(gc_grace_seconds) if gc_grace_seconds >= 0 => Some(now_secs - gc_grace_seconds),
_ => None,
},
}
}
#[cfg(feature = "write-support")]
pub(crate) fn compute_surviving_dropped_columns(
input_paths: Vec<PathBuf>,
schema: &TableSchema,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
purge_safe: bool,
max_purgeable_timestamp: Option<i64>,
) -> Result<std::collections::HashSet<String>> {
let mut surviving: std::collections::HashSet<String> = std::collections::HashSet::new();
let total = schema.dropped_columns.len();
let mut merger = KWayMerger::new_with_gc(input_paths, schema, gc_before_secs, now_secs)?
.with_purge_safe(purge_safe)
.with_max_purgeable_timestamp(max_purgeable_timestamp);
loop {
match merger.step()? {
MergeStep::Complete => break,
MergeStep::Partition { rows, .. } => {
for row in &rows {
if let RowData::Live { cells } = &row.row_data {
for cell in cells {
if schema.dropped_columns.contains_key(&cell.column) {
surviving.insert(cell.column.clone());
}
}
}
for cd in &row.complex_deletions {
if schema.dropped_columns.contains_key(&cd.column) {
surviving.insert(cd.column.clone());
}
}
}
if surviving.len() == total {
break; }
}
}
}
Ok(surviving)
}
pub async fn compact_sstables(
input_paths: Vec<PathBuf>,
output_dir: &std::path::Path,
schema: &TableSchema,
generation: u64,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
purge_safe: bool,
) -> Result<CompactReport> {
compact_sstables_with_registry(
input_paths,
output_dir,
schema,
generation,
gc_before_secs,
now_secs,
purge_safe,
None,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn compact_sstables_with_registry(
input_paths: Vec<PathBuf>,
output_dir: &std::path::Path,
schema: &TableSchema,
generation: u64,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
purge_safe: bool,
udt_registry: Option<&crate::schema::UdtRegistry>,
) -> Result<CompactReport> {
if input_paths.is_empty() {
return Err(Error::InvalidInput(
"compaction requires at least one input SSTable".to_string(),
));
}
let drop_set: Vec<PathBuf> = if purge_safe {
fully_expired_sstables(&input_paths, &[], gc_before_secs)
} else {
Vec::new()
};
let (merge_inputs, dropped_whole) = split_merge_and_dropped(&input_paths, drop_set);
let input_paths = merge_inputs;
let mut effective_schema = effective_compaction_schema(schema, &input_paths);
apply_udt_marshals_from_inputs(&mut effective_schema, &input_paths, udt_registry)?;
let retained_dropped = if effective_schema.dropped_columns.is_empty() {
std::collections::HashSet::new()
} else {
compute_surviving_dropped_columns(
input_paths.clone(),
&effective_schema,
gc_before_secs,
now_secs,
purge_safe,
None,
)?
};
let write_schema = effective_schema.for_compaction_output(&retained_dropped);
let merger = KWayMerger::new_with_gc_and_registry(
input_paths.clone(),
&effective_schema,
gc_before_secs,
now_secs,
udt_registry.cloned(),
)?
.with_purge_safe(purge_safe);
let repair_state = classify_inputs(&input_paths)?;
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
output_dir.to_path_buf(),
generation,
&write_schema,
)?;
writer.set_repair_state(
repair_state.repaired_at,
repair_state.pending_repair,
repair_state.is_transient,
);
writer.mark_compaction_output();
let (baseline_min_ts, mut baseline_min_ldt, baseline_min_ttl) =
compute_baseline_min(&input_paths);
if now_secs.is_some() {
if let Some(floor) = compute_expiry_ttl_ldt_floor(&input_paths) {
baseline_min_ldt = baseline_min_ldt.min(floor);
}
}
writer.pre_seed_encoding_baselines(baseline_min_ts, baseline_min_ldt, baseline_min_ttl);
let mut stats = merger.merge(&mut writer)?;
let output = writer.finish().await?;
fully_expired::reclaim_dropped_whole(&dropped_whole, &[], |dropped| {
if let Err(e) =
crate::storage::write_engine::WriteEngine::delete_sstable_files_static(dropped)
{
tracing::warn!(
"Failed to delete dropped-whole compaction input {:?}: {} \
(output is valid; leftover is an invisible orphan)",
dropped,
e
);
}
});
stats.dropped_whole = dropped_whole;
Ok(CompactReport { output, stats })
}
#[cfg(feature = "write-support")]
#[derive(Debug, Default, Clone, Copy)]
struct PurgeCounts {
cell_tombstones: u64,
row_tombstones: u64,
range_tombstones: u64,
complex_deletions: u64,
partition_tombstones: u64,
suppressed: u64,
emitted: u64,
}
#[cfg(feature = "write-support")]
impl PurgeCounts {
fn total(self) -> u64 {
self.cell_tombstones
.saturating_add(self.row_tombstones)
.saturating_add(self.range_tombstones)
.saturating_add(self.complex_deletions)
.saturating_add(self.partition_tombstones)
}
}
#[cfg(feature = "write-support")]
impl KWayMerger {
pub fn new(input_paths: Vec<PathBuf>, schema: &TableSchema) -> Result<Self> {
Self::new_with_gc(input_paths, schema, None, None)
}
pub fn new_cancellable(
input_paths: Vec<PathBuf>,
schema: &TableSchema,
scan_cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
Self::new_with_gc_and_registry_cancellable(
input_paths,
schema,
None,
None,
None,
scan_cancel,
)
}
#[tracing::instrument(name = "merger.new", level = "debug", skip(input_paths, schema, gc_before_secs, now_secs), fields(inputs = input_paths.len()))]
pub fn new_with_gc(
input_paths: Vec<PathBuf>,
schema: &TableSchema,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
) -> Result<Self> {
Self::new_with_gc_and_registry(input_paths, schema, gc_before_secs, now_secs, None)
}
#[tracing::instrument(name = "merger.new_registry", level = "debug", skip(input_paths, schema, gc_before_secs, now_secs, udt_registry), fields(inputs = input_paths.len()))]
pub fn new_with_gc_and_registry(
input_paths: Vec<PathBuf>,
schema: &TableSchema,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
udt_registry: Option<crate::schema::UdtRegistry>,
) -> Result<Self> {
Self::new_with_gc_and_registry_cancellable(
input_paths,
schema,
gc_before_secs,
now_secs,
udt_registry,
crate::storage::scan_cancel::ScanCancel::default(),
)
}
pub fn new_with_gc_and_registry_cancellable(
input_paths: Vec<PathBuf>,
schema: &TableSchema,
gc_before_secs: Option<i64>,
now_secs: Option<i64>,
udt_registry: Option<crate::schema::UdtRegistry>,
scan_cancel: crate::storage::scan_cancel::ScanCancel,
) -> Result<Self> {
if input_paths.is_empty() {
return Err(Error::InvalidInput(
"K-way merge requires at least one input file".to_string(),
));
}
schema.validate_dropped_columns()?;
let mut runs = Vec::with_capacity(input_paths.len());
for (run_index, path) in input_paths.iter().enumerate() {
let adapter = SSTableRowIteratorAdapter::open(
path,
run_index,
schema,
udt_registry.clone(),
scan_cancel.clone(),
)?;
runs.push(RunReader::new(Box::new(adapter)));
}
let heap = BinaryHeap::new();
Ok(Self {
runs,
heap,
current_partition: None,
schema: schema.clone(),
schema_arc: std::sync::Arc::new(schema.clone()),
gc_before_secs,
now_secs,
purge_safe: false,
max_purgeable_timestamp: None,
})
}
pub fn with_purge_safe(mut self, purge_safe: bool) -> Self {
self.purge_safe = purge_safe;
self
}
pub fn with_max_purgeable_timestamp(mut self, max_purgeable_timestamp: Option<i64>) -> Self {
self.max_purgeable_timestamp = max_purgeable_timestamp;
self
}
pub fn merge(
mut self,
output_writer: &mut crate::storage::sstable::writer::SSTableWriter,
) -> Result<MergeStats> {
let start_time = Instant::now();
let mut stats = MergeStats {
input_files: self.runs.len(),
output_partitions: 0,
output_rows: 0,
bytes_written: 0,
elapsed: Duration::from_secs(0), dropped_whole: Vec::new(),
};
let decode_schema = self.schema.clone();
let write_schema = output_writer.schema().clone();
let schema_has_static = write_schema.columns.iter().any(|c| c.is_static);
let mut stream = StreamingMerger::new(&mut self);
'partitions: loop {
let mut partition_tombstone: Option<PartitionTombstone> = None;
let mut range_tombstones: Vec<RangeTombstone> = Vec::new();
let mut static_tracker =
crate::storage::sstable::writer::data_writer::StaticOpsTracker::new();
let mut static_first_ts: i64 = 0;
let mut saw_carrier_or_static = false;
let mut row_count: u64 = 0;
let mut buffered_rows: Vec<crate::storage::write_engine::mutation::Mutation> =
Vec::new();
let mut partition_stats = crate::storage::sstable::writer::StatisticsMetadata::new();
loop {
match stream.step_streaming()? {
StreamingStep::ClusterGroup { key: _, row } => {
if row.is_metadata_only_no_op() {
continue;
}
let mutation = Self::merge_entry_to_mutation(*row, &decode_schema)?;
crate::storage::sstable::writer::stats_fold::fold_mutation_stats(
&mut partition_stats,
&mutation,
);
let is_partition_only = mutation.operations.is_empty()
&& mutation.partition_tombstone.is_some()
&& mutation.row_tombstone.is_none()
&& mutation.range_tombstones.is_empty();
let is_range_only = mutation.operations.is_empty()
&& mutation.partition_tombstone.is_none()
&& mutation.row_tombstone.is_none()
&& !mutation.range_tombstones.is_empty();
if is_partition_only {
partition_tombstone = mutation.partition_tombstone;
saw_carrier_or_static = true;
continue;
}
if is_range_only {
range_tombstones.extend(mutation.range_tombstones.iter().cloned());
saw_carrier_or_static = true;
continue;
}
if mutation.clustering_key.is_none() && schema_has_static {
if !saw_carrier_or_static {
static_first_ts = mutation.timestamp_micros;
}
static_tracker.feed(&mutation, &write_schema, None);
saw_carrier_or_static = true;
row_count += 1;
continue;
}
buffered_rows.push(mutation);
row_count += 1;
}
StreamingStep::PartitionEnd { key } => {
if !buffered_rows.is_empty() || saw_carrier_or_static {
let mut session = output_writer.begin_partition_incremental(
&key,
partition_tombstone.as_ref(),
&range_tombstones,
)?;
if schema_has_static {
let merged = std::mem::take(&mut static_tracker).finish();
session.feed_static_row(&merged, static_first_ts, &write_schema)?;
}
for mutation in &buffered_rows {
session.feed_row(mutation, &write_schema)?;
}
let (offset, blocks, emit) = session.finish(&write_schema)?;
output_writer.complete_partition_incremental(
&key,
partition_tombstone.as_ref(),
offset,
&blocks,
emit,
&partition_stats,
)?;
stats.output_partitions += 1;
stats.output_rows += row_count;
}
continue 'partitions;
}
StreamingStep::Complete => break 'partitions,
}
}
}
stats.bytes_written = output_writer.data_bytes_written();
stats.elapsed = start_time.elapsed();
Ok(stats)
}
#[tracing::instrument(name = "merger.step", level = "debug", skip(self))]
pub fn step(&mut self) -> Result<MergeStep> {
if self.heap.is_empty() && self.current_partition.is_none() {
self.initialize_heap()?;
}
if self.heap.is_empty() {
return Ok(MergeStep::Complete);
}
let mut partition_rows = Vec::new();
let mut partition_key: Option<DecoratedKey> = None;
while let Some(Reverse(wrapped)) = self.heap.peek() {
if let Some(ref current_key) = partition_key {
if &wrapped.entry.key != current_key {
break;
}
} else {
partition_key = Some(wrapped.entry.key.clone());
}
let Reverse(wrapped) = self
.heap
.pop()
.ok_or_else(|| Error::InvalidInput("Merge heap unexpectedly empty".to_string()))?;
let entry = wrapped.entry;
let run_index = entry.run_index;
partition_rows.push(entry);
self.refill_heap(run_index)?;
}
if let Some(key) = partition_key {
let merged_rows = self.merge_partition_rows(partition_rows)?;
Ok(MergeStep::Partition {
key,
rows: merged_rows,
})
} else {
Ok(MergeStep::Complete)
}
}
fn initialize_heap(&mut self) -> Result<()> {
for run_index in 0..self.runs.len() {
self.refill_heap(run_index)?;
}
Ok(())
}
fn refill_heap(&mut self, run_index: usize) -> Result<()> {
if run_index >= self.runs.len() {
return Ok(());
}
let run = &mut self.runs[run_index];
if !run.is_exhausted() {
if let Some(entry) = run.advance()? {
self.heap
.push(Reverse(schema_order::SchemaOrderedEntry::new(
entry,
self.schema_arc.clone(),
)));
}
}
Ok(())
}
fn effective_gc_settings(&self) -> (Option<i64>, i64) {
if self.purge_safe {
(self.gc_before_secs, i64::MAX)
} else if let Some(bound) = self.max_purgeable_timestamp {
(self.gc_before_secs, bound)
} else {
(None, i64::MIN)
}
}
fn merge_partition_rows(&self, rows: Vec<MergeEntry>) -> Result<Vec<MergeEntry>> {
use std::collections::BTreeMap;
let mut purges = PurgeCounts::default();
let (effective_gc_before, max_purgeable_timestamp) = self.effective_gc_settings();
let carriers::PartitionCarriers {
mut range_tombstones,
max_partition_deletion,
partition_delete_key,
} = carriers::scan_partition_carriers(&rows);
let mut clustered_rows: BTreeMap<Option<ClusteringKey>, Vec<MergeEntry>> = BTreeMap::new();
let rows_in = rows.len() as u64;
for row in rows {
if row.is_partition_delete_carrier() && row.partition_deletion.is_some() {
continue;
}
if carriers::is_range_marker_carrier(&row) {
continue;
}
clustered_rows
.entry(row.clustering_key.clone())
.or_default()
.push(row);
}
Self::coalesce_range_tombstones(&mut range_tombstones, &self.schema);
let mut merged = Vec::new();
for (ck, cluster_rows) in clustered_rows {
if let Some(entry) = Self::reconcile_cluster_with_overlap_counted(
ck,
cluster_rows,
&self.schema.dropped_columns,
effective_gc_before,
max_purgeable_timestamp,
self.now_secs,
&mut purges,
) {
if let Some(shadowed) =
Self::apply_range_shadowing(entry, &range_tombstones, &self.schema)
{
if let Some(survivor) =
Self::apply_partition_shadowing(shadowed, max_partition_deletion)
{
merged.push(survivor);
}
}
}
}
if let Some((pmfda, _)) = max_partition_deletion {
range_tombstones.retain(|(_, rt)| rt.deletion_time > pmfda);
}
for (key, rt) in range_tombstones {
if let Some(gc_before) = effective_gc_before {
if i64::from(rt.local_deletion_time as u32) < gc_before
&& rt.deletion_time < max_purgeable_timestamp
{
purges.range_tombstones += 1;
continue;
}
}
purges.emitted += 1;
merged.push(
MergeEntry::new(
usize::MAX,
key,
None,
rt.deletion_time,
RowData::Live { cells: Vec::new() },
)
.with_range_deletion(rt),
);
}
if let (Some((pmfda, pldt)), Some(key)) = (max_partition_deletion, partition_delete_key) {
let purge = match effective_gc_before {
Some(gc_before) => {
i64::from(pldt as u32) < gc_before && pmfda < max_purgeable_timestamp
}
None => false,
};
if purge {
purges.partition_tombstones += 1;
} else {
purges.emitted += 1;
merged.push(
MergeEntry::new(
usize::MAX,
key,
None,
pmfda,
RowData::Tombstone {
deletion_time: pmfda,
local_deletion_time: pldt,
},
)
.with_partition_deletion((pmfda, pldt)),
);
}
}
merged.sort_by(|a, b| match (&a.clustering_key, &b.clustering_key) {
(None, None) => Ordering::Equal,
(None, Some(_)) => Ordering::Less,
(Some(_), None) => Ordering::Greater,
(Some(ck_a), Some(ck_b)) => {
ck_a.compare(ck_b, &self.schema).unwrap_or_else(|e| {
tracing::warn!(
"Schema-aware clustering key comparison failed, using fallback: {}",
e
);
ck_a.cmp(ck_b)
})
}
});
let purged = purges.total();
if purged > 0 {
crate::observability::add_counter(
crate::observability::catalog::COMPACTION_TOMBSTONES_PURGED,
purged,
&[],
);
}
if purges.suppressed > 0 {
crate::observability::add_counter(
crate::observability::catalog::COMPACTION_TOMBSTONES_SUPPRESSED,
purges.suppressed,
&[],
);
}
if purges.emitted > 0 {
crate::observability::add_counter(
crate::observability::catalog::COMPACTION_TOMBSTONES_EMITTED,
purges.emitted,
&[],
);
}
if rows_in > 0 {
crate::observability::add_counter(
crate::observability::catalog::MERGE_ROWS_IN,
rows_in,
&[],
);
crate::observability::add_counter(
crate::observability::catalog::MERGE_ROWS_OUT,
merged.len() as u64,
&[],
);
}
Ok(merged)
}
fn coalesce_range_tombstones(
rts: &mut Vec<(DecoratedKey, RangeTombstone)>,
schema: &TableSchema,
) {
let mut groups: Vec<(DecoratedKey, Vec<RangeTombstone>)> = Vec::new();
for (key, rt) in rts.drain(..) {
if let Some((_, ranges)) = groups.iter_mut().find(|(k, _)| k.key == key.key) {
ranges.push(rt);
} else {
groups.push((key, vec![rt]));
}
}
let mut out: Vec<(DecoratedKey, RangeTombstone)> = Vec::new();
for (key, ranges) in groups {
for rt in Self::coalesce_partition_range_tombstones(ranges, schema) {
out.push((key.clone(), rt));
}
}
*rts = out;
}
fn coalesce_partition_range_tombstones(
ranges: Vec<RangeTombstone>,
schema: &TableSchema,
) -> Vec<RangeTombstone> {
if ranges.len() <= 1 {
return ranges;
}
let items: Vec<(RangeCut, RangeCut, i64, i32)> = ranges
.iter()
.map(|rt| {
(
Self::range_start_cut(&rt.start),
Self::range_end_cut(&rt.end),
rt.deletion_time,
rt.local_deletion_time,
)
})
.collect();
let mut cuts: Vec<RangeCut> = Vec::with_capacity(items.len() * 2);
for (s, e, _, _) in &items {
cuts.push(s.clone());
cuts.push(e.clone());
}
cuts.sort_by(|a, b| Self::cut_cmp(a, b, schema));
cuts.dedup_by(|a, b| Self::cut_cmp(a, b, schema) == Ordering::Equal);
let mut segs: Vec<(RangeCut, RangeCut, i64, i32)> = Vec::new();
for window in cuts.windows(2) {
let (lo, hi) = (&window[0], &window[1]);
let mut best: Option<(i64, i32)> = None;
for (s, e, mfda, ldt) in &items {
let covers = Self::cut_cmp(s, lo, schema) != Ordering::Greater
&& Self::cut_cmp(hi, e, schema) != Ordering::Greater;
if !covers {
continue;
}
best = Some(match best {
Some((bm, bl)) if bm > *mfda || (bm == *mfda && bl >= *ldt) => (bm, bl),
_ => (*mfda, *ldt),
});
}
if let Some((mfda, ldt)) = best {
segs.push((lo.clone(), hi.clone(), mfda, ldt));
}
}
let mut merged: Vec<(RangeCut, RangeCut, i64, i32)> = Vec::new();
for seg in segs {
if let Some(last) = merged.last_mut() {
if last.2 == seg.2
&& last.3 == seg.3
&& Self::cut_cmp(&last.1, &seg.0, schema) == Ordering::Equal
{
last.1 = seg.1;
continue;
}
}
merged.push(seg);
}
merged
.into_iter()
.map(|(lo, hi, mfda, ldt)| RangeTombstone {
start: Self::cut_to_start_bound(lo),
end: Self::cut_to_end_bound(hi),
deletion_time: mfda,
local_deletion_time: ldt,
})
.collect()
}
fn cut_cmp(a: &RangeCut, b: &RangeCut, schema: &TableSchema) -> Ordering {
match (a, b) {
(RangeCut::Bottom, RangeCut::Bottom) => Ordering::Equal,
(RangeCut::Bottom, _) => Ordering::Less,
(_, RangeCut::Bottom) => Ordering::Greater,
(RangeCut::Top, RangeCut::Top) => Ordering::Equal,
(RangeCut::Top, _) => Ordering::Greater,
(_, RangeCut::Top) => Ordering::Less,
(RangeCut::At { key: ka, after: aa }, RangeCut::At { key: kb, after: ab }) => {
let l = ka.columns.len().min(kb.columns.len());
let ta = ClusteringKey {
columns: ka.columns[..l].to_vec(),
};
let tb = ClusteringKey {
columns: kb.columns[..l].to_vec(),
};
let ord = ta.compare(&tb, schema).unwrap_or_else(|_| ta.cmp(&tb));
if ord != Ordering::Equal {
return ord;
}
match ka.columns.len().cmp(&kb.columns.len()) {
Ordering::Equal => aa.cmp(ab),
Ordering::Less => {
if *aa {
Ordering::Greater
} else {
Ordering::Less
}
}
Ordering::Greater => {
if *ab {
Ordering::Less
} else {
Ordering::Greater
}
}
}
}
}
}
fn range_start_cut(
bound: &crate::storage::write_engine::mutation::ClusteringBound,
) -> RangeCut {
use crate::storage::write_engine::mutation::ClusteringBound;
match bound {
ClusteringBound::Inclusive(ck) => RangeCut::At {
key: ck.clone(),
after: false,
},
ClusteringBound::Exclusive(ck) => RangeCut::At {
key: ck.clone(),
after: true,
},
ClusteringBound::Bottom => RangeCut::Bottom,
ClusteringBound::Top => RangeCut::Top,
}
}
fn range_end_cut(bound: &crate::storage::write_engine::mutation::ClusteringBound) -> RangeCut {
use crate::storage::write_engine::mutation::ClusteringBound;
match bound {
ClusteringBound::Inclusive(ck) => RangeCut::At {
key: ck.clone(),
after: true,
},
ClusteringBound::Exclusive(ck) => RangeCut::At {
key: ck.clone(),
after: false,
},
ClusteringBound::Top => RangeCut::Top,
ClusteringBound::Bottom => RangeCut::Bottom,
}
}
fn cut_to_start_bound(
cut: RangeCut,
) -> crate::storage::write_engine::mutation::ClusteringBound {
use crate::storage::write_engine::mutation::ClusteringBound;
match cut {
RangeCut::Bottom => ClusteringBound::Bottom,
RangeCut::At { key, after: false } => ClusteringBound::Inclusive(key),
RangeCut::At { key, after: true } => ClusteringBound::Exclusive(key),
RangeCut::Top => ClusteringBound::Top,
}
}
fn cut_to_end_bound(cut: RangeCut) -> crate::storage::write_engine::mutation::ClusteringBound {
use crate::storage::write_engine::mutation::ClusteringBound;
match cut {
RangeCut::Top => ClusteringBound::Top,
RangeCut::At { key, after: true } => ClusteringBound::Inclusive(key),
RangeCut::At { key, after: false } => ClusteringBound::Exclusive(key),
RangeCut::Bottom => ClusteringBound::Bottom,
}
}
fn apply_partition_shadowing(
entry: MergeEntry,
max_partition_deletion: Option<(i64, i32)>,
) -> Option<MergeEntry> {
let Some((pmfda, _)) = max_partition_deletion else {
return Some(entry);
};
let surviving_complex: Vec<ComplexDeletion> = entry
.complex_deletions
.into_iter()
.filter(|cd| cd.marked_for_delete_at > pmfda)
.collect();
let surviving_row_del = entry.row_deletion.filter(|(dt, _)| *dt > pmfda);
match entry.row_data {
RowData::Tombstone {
deletion_time,
local_deletion_time,
} => {
if deletion_time > pmfda {
let mut rebuilt = MergeEntry::new(
entry.run_index,
entry.key,
entry.clustering_key,
deletion_time,
RowData::Tombstone {
deletion_time,
local_deletion_time,
},
);
if !surviving_complex.is_empty() {
rebuilt = rebuilt.with_complex_deletions(surviving_complex);
}
Some(rebuilt)
} else if !surviving_complex.is_empty() {
Some(
MergeEntry::new(
entry.run_index,
entry.key,
entry.clustering_key,
entry.timestamp,
RowData::Live { cells: Vec::new() },
)
.with_complex_deletions(surviving_complex),
)
} else {
None
}
}
RowData::Live { cells } => {
let ck_names: std::collections::HashSet<String> = entry
.clustering_key
.as_ref()
.map(|ck| ck.columns.iter().map(|(n, _)| n.clone()).collect())
.unwrap_or_default();
let is_data = |c: &CellData| !ck_names.contains(&c.column);
let kept: Vec<CellData> = cells
.into_iter()
.filter(|c| !is_data(c) || c.timestamp > pmfda)
.collect();
let has_data = kept.iter().any(is_data);
let marker_live = entry.timestamp > pmfda;
if !has_data && !marker_live {
if let Some((dt, ldt)) = surviving_row_del {
return Some(MergeEntry::new(
entry.run_index,
entry.key,
entry.clustering_key,
dt,
RowData::Tombstone {
deletion_time: dt,
local_deletion_time: ldt,
},
));
}
if !surviving_complex.is_empty() {
return Some(
MergeEntry::new(
entry.run_index,
entry.key,
entry.clustering_key,
entry.timestamp,
RowData::Live { cells: Vec::new() },
)
.with_complex_deletions(surviving_complex),
);
}
return None;
}
let row_ts = if has_data {
kept.iter()
.filter(|c| is_data(c))
.map(|c| c.timestamp)
.max()
.unwrap_or(entry.timestamp)
} else {
entry.timestamp
};
let mut rebuilt = MergeEntry::new(
entry.run_index,
entry.key,
entry.clustering_key,
row_ts,
RowData::Live { cells: kept },
);
if !surviving_complex.is_empty() {
rebuilt = rebuilt.with_complex_deletions(surviving_complex);
}
if let Some((dt, ldt)) = surviving_row_del {
rebuilt = rebuilt.with_row_deletion(dt, ldt);
}
Some(rebuilt)
}
}
}
fn range_tombstone_covers_ck(
ck: &ClusteringKey,
rt: &RangeTombstone,
schema: &TableSchema,
) -> bool {
#[cfg(test)]
crate::storage::sstable::work_counters::range_coverage_scope::record();
use crate::storage::write_engine::mutation::ClusteringBound;
let cmp = |bound: &ClusteringKey| -> Ordering {
let n = bound.columns.len();
let truncated = ClusteringKey {
columns: ck.columns.iter().take(n).cloned().collect(),
};
truncated
.compare(bound, schema)
.unwrap_or_else(|_| truncated.cmp(bound))
};
let after_start = match &rt.start {
ClusteringBound::Inclusive(b) => cmp(b) != Ordering::Less,
ClusteringBound::Exclusive(b) => cmp(b) == Ordering::Greater,
ClusteringBound::Bottom => true,
ClusteringBound::Top => false,
};
let before_end = match &rt.end {
ClusteringBound::Inclusive(b) => cmp(b) != Ordering::Greater,
ClusteringBound::Exclusive(b) => cmp(b) == Ordering::Less,
ClusteringBound::Top => true,
ClusteringBound::Bottom => false,
};
after_start && before_end
}
fn range_end_before_ck(ck: &ClusteringKey, rt: &RangeTombstone, schema: &TableSchema) -> bool {
use crate::storage::write_engine::mutation::ClusteringBound;
let cmp = |bound: &ClusteringKey| -> Ordering {
let n = bound.columns.len();
let truncated = ClusteringKey {
columns: ck.columns.iter().take(n).cloned().collect(),
};
truncated
.compare(bound, schema)
.unwrap_or_else(|_| truncated.cmp(bound))
};
match &rt.end {
ClusteringBound::Inclusive(b) => cmp(b) == Ordering::Greater,
ClusteringBound::Exclusive(b) => cmp(b) != Ordering::Less,
ClusteringBound::Top => false,
ClusteringBound::Bottom => true,
}
}
fn apply_range_shadowing(
entry: MergeEntry,
range_tombstones: &[(DecoratedKey, RangeTombstone)],
schema: &TableSchema,
) -> Option<MergeEntry> {
if range_tombstones.is_empty() {
return Some(entry);
}
let Some(ck) = entry.clustering_key.clone() else {
return Some(entry);
};
let single_partition = match (range_tombstones.first(), range_tombstones.last()) {
(Some((first, _)), Some((last, _))) => {
first.key == entry.key.key && last.key == entry.key.key
}
_ => false,
};
let floor = if single_partition {
let idx = range_tombstones
.partition_point(|(_, rt)| Self::range_end_before_ck(&ck, rt, schema));
range_tombstones.get(idx).and_then(|(key, rt)| {
(key.key == entry.key.key && Self::range_tombstone_covers_ck(&ck, rt, schema))
.then_some(rt.deletion_time)
})
} else {
range_tombstones
.iter()
.filter(|(key, rt)| {
key.key == entry.key.key && Self::range_tombstone_covers_ck(&ck, rt, schema)
})
.map(|(_, rt)| rt.deletion_time)
.max()
};
let Some(floor) = floor else {
return Some(entry);
};
let surviving_complex: Vec<ComplexDeletion> = entry
.complex_deletions
.into_iter()
.filter(|cd| cd.marked_for_delete_at > floor)
.collect();
match entry.row_data {
RowData::Tombstone {
deletion_time,
local_deletion_time,
} => {
if deletion_time > floor {
let mut rebuilt = MergeEntry::new(
entry.run_index,
entry.key,
Some(ck),
deletion_time,
RowData::Tombstone {
deletion_time,
local_deletion_time,
},
);
if !surviving_complex.is_empty() {
rebuilt = rebuilt.with_complex_deletions(surviving_complex);
}
Some(rebuilt)
} else if !surviving_complex.is_empty() {
Some(
MergeEntry::new(
entry.run_index,
entry.key,
Some(ck),
entry.timestamp,
RowData::Live { cells: Vec::new() },
)
.with_complex_deletions(surviving_complex),
)
} else {
None
}
}
RowData::Live { cells } => {
let ck_names: std::collections::HashSet<&str> =
ck.columns.iter().map(|(n, _)| n.as_str()).collect();
let is_data = |c: &CellData| !ck_names.contains(c.column.as_str());
let kept: Vec<CellData> = cells
.into_iter()
.filter(|c| !is_data(c) || c.timestamp > floor)
.collect();
let has_data = kept.iter().any(is_data);
let surviving_row_del = entry.row_deletion.filter(|(dt, _)| *dt > floor);
let marker_live = entry.timestamp > floor;
if !has_data && !marker_live {
if let Some((dt, ldt)) = surviving_row_del {
return Some(MergeEntry::new(
entry.run_index,
entry.key,
Some(ck),
dt,
RowData::Tombstone {
deletion_time: dt,
local_deletion_time: ldt,
},
));
}
if !surviving_complex.is_empty() {
return Some(
MergeEntry::new(
entry.run_index,
entry.key,
Some(ck),
entry.timestamp,
RowData::Live { cells: Vec::new() },
)
.with_complex_deletions(surviving_complex),
);
}
return None;
}
let row_ts = if has_data {
kept.iter()
.filter(|c| is_data(c))
.map(|c| c.timestamp)
.max()
.unwrap_or(entry.timestamp)
} else {
entry.timestamp
};
let mut rebuilt = MergeEntry::new(
entry.run_index,
entry.key,
Some(ck),
row_ts,
RowData::Live { cells: kept },
);
if !surviving_complex.is_empty() {
rebuilt = rebuilt.with_complex_deletions(surviving_complex);
}
if let Some((dt, ldt)) = surviving_row_del {
rebuilt = rebuilt.with_row_deletion(dt, ldt);
}
Some(rebuilt)
}
}
}
fn cell_effective_ldt(cell: &CellData) -> Option<i32> {
if let Some(ldt) = cell.local_deletion_time {
return Some(ldt);
}
if let crate::types::Value::Tombstone(ref info) = cell.value {
if info.tombstone_type == crate::types::TombstoneType::CellTombstone
&& info.local_deletion_time != 0
{
return Some(info.local_deletion_time as i32);
}
}
None
}
fn is_cell_tombstone(cell: &CellData) -> bool {
matches!(
cell.value,
crate::types::Value::Tombstone(ref info)
if info.tombstone_type == crate::types::TombstoneType::CellTombstone
)
|| cell.is_deleted
}
#[cfg(test)]
fn reconcile_cluster(
clustering_key: Option<ClusteringKey>,
cluster_rows: Vec<MergeEntry>,
dropped_columns: &std::collections::HashMap<String, i64>,
gc_before_secs: Option<i64>,
) -> Option<MergeEntry> {
Self::reconcile_cluster_with_overlap(
clustering_key,
cluster_rows,
dropped_columns,
gc_before_secs,
i64::MAX,
)
}
#[cfg(test)]
fn reconcile_cluster_with_overlap(
clustering_key: Option<ClusteringKey>,
cluster_rows: Vec<MergeEntry>,
dropped_columns: &std::collections::HashMap<String, i64>,
gc_before_secs: Option<i64>,
max_purgeable_timestamp: i64,
) -> Option<MergeEntry> {
let mut sink = PurgeCounts::default();
Self::reconcile_cluster_with_overlap_counted(
clustering_key,
cluster_rows,
dropped_columns,
gc_before_secs,
max_purgeable_timestamp,
None,
&mut sink,
)
}
fn reconcile_cluster_with_overlap_counted(
clustering_key: Option<ClusteringKey>,
cluster_rows: Vec<MergeEntry>,
dropped_columns: &std::collections::HashMap<String, i64>,
gc_before_secs: Option<i64>,
max_purgeable_timestamp: i64,
now_secs: Option<i64>,
purges: &mut PurgeCounts,
) -> Option<MergeEntry> {
let mut state = reconcile::ReconcileState::new(clustering_key);
state.fold_row_deletions(&cluster_rows);
state.resolve_cell_winners(&cluster_rows);
if !state.has_key() {
return None;
}
state.apply_complex_deletions();
state.shadow_by_row_deletion(purges);
state.filter_dropped_columns(dropped_columns);
state.expire_ttl_cells(now_secs);
state.purge_gc_grace(gc_before_secs, max_purgeable_timestamp, purges);
state.build(purges)
}
fn cells_to_cell_operations(
cells: Vec<CellData>,
) -> Vec<crate::storage::write_engine::mutation::CellOperation> {
use crate::storage::write_engine::mutation::CellOperation;
use crate::types::{TombstoneType, Value};
cells
.into_iter()
.map(|cell| {
if cell.is_complex_element {
let value = if cell.is_deleted || cell.has_empty_value {
None
} else {
Some(cell.value)
};
return CellOperation::WriteComplexElement {
column: cell.column,
cell_path: cell.cell_path.unwrap_or_default(),
value,
timestamp_micros: cell.timestamp,
ttl_seconds: cell.ttl,
local_deletion_time: cell.local_deletion_time,
is_deleted: cell.is_deleted,
};
}
if let Value::Tombstone(ref info) = cell.value {
if info.tombstone_type == TombstoneType::CellTombstone {
let preserved_ldt = match info.local_deletion_time {
0 => None,
ldt => Some(ldt as i32),
};
return CellOperation::Delete {
column: cell.column,
local_deletion_time: preserved_ldt,
};
}
}
if let Some(ttl) = cell.ttl {
CellOperation::WriteWithTtl {
column: cell.column,
value: cell.value,
ttl_seconds: ttl,
local_deletion_time: cell.local_deletion_time,
}
} else {
CellOperation::Write {
column: cell.column,
value: cell.value,
}
}
})
.collect()
}
pub(crate) fn merge_entry_to_mutation(
entry: MergeEntry,
schema: &TableSchema,
) -> Result<crate::storage::write_engine::mutation::Mutation> {
use crate::storage::write_engine::mutation::{
CellOperation, Mutation, PartitionKey, TableId,
};
let partition_key = PartitionKey::from_bytes(&entry.key.key, schema)?;
let table_id = TableId::new(&schema.keyspace, &schema.table);
if let Some((deletion_time, local_deletion_time)) = entry.partition_deletion {
let mutation = Mutation::new(
table_id,
partition_key,
None,
Vec::new(),
deletion_time,
None,
);
let mut mutation = mutation;
mutation.partition_tombstone =
Some(crate::storage::write_engine::mutation::PartitionTombstone {
deletion_time,
local_deletion_time,
});
return Ok(mutation);
}
let range_tombstone = entry.range_deletion.clone();
let row_tombstone_ldt = match &entry.row_data {
RowData::Tombstone {
local_deletion_time,
..
} if *local_deletion_time != 0 => Some(*local_deletion_time),
_ => None,
};
let coexisting_row_tombstone = match &entry.row_data {
RowData::Live { .. } => entry.row_deletion,
RowData::Tombstone { .. } => None,
};
let row_del = match &entry.row_data {
RowData::Tombstone { deletion_time, .. } => Some(*deletion_time),
RowData::Live { .. } => None,
};
let cell_write_timestamps: Option<std::collections::HashMap<String, i64>> =
match &entry.row_data {
RowData::Live { cells } => {
let map: std::collections::HashMap<String, i64> = cells
.iter()
.filter(|c| !c.is_complex_element && c.timestamp != entry.timestamp)
.map(|c| (c.column.clone(), c.timestamp))
.collect();
(!map.is_empty()).then_some(map)
}
RowData::Tombstone { .. } => None,
};
let mut operations = match entry.row_data {
RowData::Live { cells } => Self::cells_to_cell_operations(cells),
RowData::Tombstone { .. } => vec![CellOperation::DeleteRow],
};
for cd in entry.complex_deletions {
let strictly_supersedes_row_tombstone =
row_del.is_none_or(|rd| cd.marked_for_delete_at > rd);
if strictly_supersedes_row_tombstone {
operations.push(CellOperation::ComplexDeletion {
column: cd.column,
marked_for_delete_at: cd.marked_for_delete_at,
local_deletion_time: cd.local_deletion_time,
});
}
}
let mutation = Mutation::new(
table_id,
partition_key,
entry.clustering_key,
operations,
entry.timestamp,
None,
);
let mutation = match row_tombstone_ldt {
Some(ldt) => mutation.with_local_deletion_time(ldt),
None => mutation,
};
let mut mutation = match coexisting_row_tombstone {
Some((deletion_time, ldt)) => mutation.with_row_tombstone(deletion_time, ldt),
None => mutation,
};
if let Some(rt) = range_tombstone {
mutation.range_tombstones.push(rt);
}
mutation.cell_write_timestamps = cell_write_timestamps;
Ok(mutation)
}
}
#[cfg(all(test, feature = "write-support"))]
mod tests {
use super::*;
use crate::storage::write_engine::mutation::DecoratedKey;
#[test]
fn test_merge_entry_ordering_by_token() {
let entry1 = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000,
RowData::Live { cells: vec![] },
);
let entry2 = MergeEntry::new(
0,
DecoratedKey::new(200, vec![1, 2, 3]),
None,
1000,
RowData::Live { cells: vec![] },
);
assert!(entry1 < entry2);
assert!(entry2 > entry1);
}
#[test]
fn test_merge_entry_ordering_by_key_bytes() {
let entry1 = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000,
RowData::Live { cells: vec![] },
);
let entry2 = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 4]),
None,
1000,
RowData::Live { cells: vec![] },
);
assert!(entry1 < entry2);
assert!(entry2 > entry1);
}
#[test]
fn test_merge_entry_ordering_by_run_index() {
let entry1 = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000,
RowData::Live { cells: vec![] },
);
let entry2 = MergeEntry::new(
1,
DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000,
RowData::Live { cells: vec![] },
);
assert!(entry1 < entry2);
assert!(entry2 > entry1);
}
#[test]
fn test_merge_entry_min_heap() {
use std::cmp::Reverse;
use std::collections::BinaryHeap;
let mut heap: BinaryHeap<Reverse<MergeEntry>> = BinaryHeap::new();
let entry3 = MergeEntry::new(
0,
DecoratedKey::new(300, vec![3]),
None,
1000,
RowData::Live { cells: vec![] },
);
let entry1 = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1]),
None,
1000,
RowData::Live { cells: vec![] },
);
let entry2 = MergeEntry::new(
0,
DecoratedKey::new(200, vec![2]),
None,
1000,
RowData::Live { cells: vec![] },
);
heap.push(Reverse(entry3.clone()));
heap.push(Reverse(entry1.clone()));
heap.push(Reverse(entry2.clone()));
assert_eq!(heap.pop().unwrap().0.key.token, 100);
assert_eq!(heap.pop().unwrap().0.key.token, 200);
assert_eq!(heap.pop().unwrap().0.key.token, 300);
}
#[test]
fn test_row_data_variants() {
let live = RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("Alice".to_string()),
timestamp: 1000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
};
match live {
RowData::Live { cells } => {
assert_eq!(cells.len(), 1);
assert_eq!(cells[0].column, "name");
}
_ => panic!("Expected Live variant"),
}
let tombstone = RowData::Tombstone {
deletion_time: 2000,
local_deletion_time: 1000,
};
match tombstone {
RowData::Tombstone {
deletion_time,
local_deletion_time,
} => {
assert_eq!(deletion_time, 2000);
assert_eq!(local_deletion_time, 1000);
}
_ => panic!("Expected Tombstone variant"),
}
}
#[test]
fn test_cell_data_creation() {
let cell = CellData {
column: "age".to_string(),
value: Value::Integer(30),
timestamp: 1234567890,
ttl: Some(3600),
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
assert_eq!(cell.column, "age");
assert_eq!(cell.value, Value::Integer(30));
assert_eq!(cell.timestamp, 1234567890);
assert_eq!(cell.ttl, Some(3600));
}
#[test]
fn test_merge_stats_creation() {
let stats = MergeStats {
input_files: 5,
output_partitions: 1000,
output_rows: 5000,
bytes_written: 1024 * 1024,
elapsed: Duration::from_secs(10),
dropped_whole: Vec::new(),
};
assert_eq!(stats.input_files, 5);
assert_eq!(stats.output_partitions, 1000);
assert_eq!(stats.output_rows, 5000);
assert_eq!(stats.bytes_written, 1024 * 1024);
assert_eq!(stats.elapsed.as_secs(), 10);
}
#[test]
fn test_run_reader_estimate_entry_size() {
let entry = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 3, 4]),
None,
1000,
RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("Alice".to_string()),
timestamp: 1000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let size = RunReader::estimate_entry_size(&entry);
let expected_min_size = std::mem::size_of::<MergeEntry>() + 4;
assert!(size >= expected_min_size);
}
#[test]
fn test_kway_merger_empty_input() {
use crate::schema::{KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let result = KWayMerger::new(vec![], &schema);
assert!(result.is_err());
if let Err(Error::InvalidInput(msg)) = result {
assert!(msg.contains("at least one input file"));
} else {
panic!("Expected InvalidInput error");
}
}
#[test]
fn test_merge_entry_equal_timestamps_prefer_lower_run_index() {
let entry_run0 = MergeEntry::new(
0, DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000, RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("Newer".to_string()),
timestamp: 1000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let entry_run1 = MergeEntry::new(
1, DecoratedKey::new(100, vec![1, 2, 3]),
None,
1000, RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("Older".to_string()),
timestamp: 1000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
assert!(entry_run0 < entry_run1);
}
#[test]
fn test_merge_entry_tombstone() {
let tombstone_entry = MergeEntry::new(
0,
DecoratedKey::new(100, vec![1, 2, 3]),
None,
2000,
RowData::Tombstone {
deletion_time: 2000,
local_deletion_time: 1000,
},
);
match tombstone_entry.row_data {
RowData::Tombstone {
deletion_time,
local_deletion_time,
} => {
assert_eq!(deletion_time, 2000);
assert_eq!(local_deletion_time, 1000);
}
_ => panic!("Expected Tombstone"),
}
}
#[test]
fn test_real_merger_delete_wins_at_equal_timestamp() {
use crate::schema::{Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "reconcile_ks".to_string(),
table: "reconcile_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "value".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
const EQUAL_TS: i64 = 1_700_000_000_000_000;
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let live_entry = MergeEntry::new(
0,
partition_key.clone(),
None,
EQUAL_TS,
RowData::Live {
cells: vec![CellData {
column: "value".to_string(),
value: Value::text("survivor-if-buggy".to_string()),
timestamp: EQUAL_TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let tombstone_entry = MergeEntry::new(
1,
partition_key.clone(),
None,
EQUAL_TS,
RowData::Tombstone {
deletion_time: EQUAL_TS,
local_deletion_time: 2_000_000,
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![live_entry, tombstone_entry])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1, "one clustering key => one merged winner");
assert!(
matches!(merged[0].row_data, RowData::Tombstone { .. }),
"At equal timestamp the tombstone must win even though the live row is in \
the newer file (run_index 0). Got a live row => the equal-ts tiebreak \
reverted to run_index (Issue #498 regression)."
);
}
#[test]
fn test_real_merger_disjoint_columns_survive_compaction() {
use crate::schema::{ClusteringColumn, Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "disjoint_ks".to_string(),
table: "disjoint_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: Default::default(),
}],
columns: vec![
Column {
name: "name".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
Column {
name: "score".to_string(),
data_type: "int".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let ck = ClusteringKey {
columns: vec![("ck".to_string(), Value::Integer(1))],
};
let entry_a = MergeEntry::new(
1,
partition_key.clone(),
Some(ck.clone()),
100,
RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("alice".to_string()),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let entry_b = MergeEntry::new(
0,
partition_key.clone(),
Some(ck.clone()),
200,
RowData::Live {
cells: vec![CellData {
column: "score".to_string(),
value: Value::Integer(42),
timestamp: 200,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![entry_b, entry_a])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1, "one clustering key => one merged row");
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected a Live merged row, got {:?}", other),
};
let name = cells.iter().find(|c| c.column == "name");
let score = cells.iter().find(|c| c.column == "score");
assert!(
name.is_some(),
"disjoint column `name` from the older file was DROPPED — per-cell \
reconcile regression (Issue #533). Old whole-row-wins code fails here."
);
assert!(
score.is_some(),
"disjoint column `score` from the newer file is missing"
);
assert_eq!(
name.unwrap().value,
Value::text("alice".to_string()),
"`name` must carry A's value"
);
assert_eq!(
score.unwrap().value,
Value::Integer(42),
"`score` must carry B's value"
);
assert_eq!(
merged[0].timestamp, 200,
"merged row timestamp must be the max surviving cell timestamp"
);
}
#[test]
fn test_real_merger_cell_tombstone_beats_live_at_equal_timestamp() {
use crate::schema::{ClusteringColumn, Column, KeyColumn, TableSchema};
use crate::types::{TombstoneInfo, TombstoneType};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "ct_ks".to_string(),
table: "ct_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: Default::default(),
}],
columns: vec![Column {
name: "score".to_string(),
data_type: "int".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let ck = ClusteringKey {
columns: vec![("ck".to_string(), Value::Integer(1))],
};
let entry_a = MergeEntry::new(
0,
partition_key.clone(),
Some(ck.clone()),
100,
RowData::Live {
cells: vec![CellData {
column: "score".to_string(),
value: Value::Integer(42),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let entry_b = MergeEntry::new(
1,
partition_key.clone(),
Some(ck.clone()),
100,
RowData::Live {
cells: vec![CellData {
column: "score".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: 100,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: 0,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![entry_a, entry_b])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1, "one clustering key => one merged row");
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected a Live merged row, got {:?}", other),
};
let score = cells
.iter()
.find(|c| c.column == "score")
.expect("score cell must be present (as a tombstone)");
assert!(
matches!(
score.value,
Value::Tombstone(ref info) if info.tombstone_type == TombstoneType::CellTombstone
),
"at equal ts the cell tombstone must win over the live value (got {:?}) — \
a recency-only tiebreak would have kept the newer file's live 42 (#498 per cell)",
score.value
);
}
#[test]
fn test_real_merger_value_tiebreak_diverges_from_cassandra() {
use crate::schema::{Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "tiebreak_ks".to_string(),
table: "tiebreak_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "v".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
const EQUAL_TS: i64 = 1_700_000_000_000_000;
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let newer_file_value = "apple"; let older_file_value = "banana"; assert!(
older_file_value.as_bytes() > newer_file_value.as_bytes(),
"fixture invariant: the older file must hold the lexicographically GREATER \
raw value, so the two tie-break rules pick different winners"
);
let cassandra_winner = if older_file_value.as_bytes() > newer_file_value.as_bytes() {
older_file_value
} else {
newer_file_value
};
let entry_newer = MergeEntry::new(
0, partition_key.clone(),
None,
EQUAL_TS,
RowData::Live {
cells: vec![CellData {
column: "v".to_string(),
value: Value::text(newer_file_value.to_string()),
timestamp: EQUAL_TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let entry_older = MergeEntry::new(
1, partition_key.clone(),
None,
EQUAL_TS,
RowData::Live {
cells: vec![CellData {
column: "v".to_string(),
value: Value::text(older_file_value.to_string()),
timestamp: EQUAL_TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![entry_newer, entry_older])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1, "one (pk, ck) group => one merged winner");
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected a Live merged row, got {:?}", other),
};
let surviving = match &cells
.iter()
.find(|c| c.column == "v")
.expect("column `v` must survive")
.value
{
Value::Text(s) => String::from_utf8_lossy(s).into_owned(),
other => panic!("expected Text value, got {:?}", other),
};
assert_eq!(
surviving, newer_file_value,
"CQLite reconcile_cluster keeps the first-seen (newer file) cell on a \
timestamp tie; got {:?}",
surviving
);
assert_ne!(
surviving, cassandra_winner,
"EXPECTED DIVERGENCE (#4/#21): CQLite kept {:?} but Cassandra's \
Cells.resolveRegular keeps the greater raw value {:?}. If this assertion \
fails, CQLite now matches Cassandra and the finding is RESOLVED — update \
cqlite-findings-and-applicability.md and convert this into a convergence test.",
surviving, cassandra_winner
);
}
#[test]
fn test_real_merger_same_column_conflict_resolves_by_timestamp() {
use crate::schema::{Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "conflict_ks".to_string(),
table: "conflict_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "name".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
Column {
name: "extra".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let entry_a = MergeEntry::new(
1,
partition_key.clone(),
None,
100,
RowData::Live {
cells: vec![
CellData {
column: "name".to_string(),
value: Value::text("old".to_string()),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
CellData {
column: "extra".to_string(),
value: Value::text("a-only".to_string()),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let entry_b = MergeEntry::new(
0,
partition_key.clone(),
None,
200,
RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("new".to_string()),
timestamp: 200,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![entry_b, entry_a])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1);
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
let name = cells
.iter()
.find(|c| c.column == "name")
.expect("name present");
let extra = cells
.iter()
.find(|c| c.column == "extra")
.expect("extra (disjoint) must survive");
assert_eq!(
name.value,
Value::text("new".to_string()),
"same-column conflict must resolve to the higher-timestamp value"
);
assert_eq!(
extra.value,
Value::text("a-only".to_string()),
"disjoint column from the older file must survive the conflict merge"
);
}
#[test]
fn test_real_merger_row_tombstone_shadows_old_cells_keeps_new() {
use crate::schema::{Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "shadow_ks".to_string(),
table: "shadow_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "name".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
Column {
name: "score".to_string(),
data_type: "int".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let pk = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let entry_a = MergeEntry::new(
2,
pk.clone(),
None,
100,
RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("old".to_string()),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let entry_b = MergeEntry::new(
1,
pk.clone(),
None,
200,
RowData::Tombstone {
deletion_time: 200,
local_deletion_time: 0,
},
);
let entry_c = MergeEntry::new(
0,
pk.clone(),
None,
300,
RowData::Live {
cells: vec![CellData {
column: "score".to_string(),
value: Value::Integer(7),
timestamp: 300,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![entry_c, entry_b, entry_a])
.expect("merge must not fail");
assert_eq!(merged.len(), 1);
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!(
"expected Live (score survives the tombstone), got {:?}",
other
),
};
assert!(
cells.iter().all(|c| c.column != "name"),
"`name` (ts=100 <= row_del=200) must be shadowed by the row tombstone"
);
let score = cells
.iter()
.find(|c| c.column == "score")
.expect("`score` (ts=300 > row_del=200) must survive the row tombstone");
assert_eq!(score.value, Value::Integer(7));
}
#[test]
fn test_real_merger_row_tombstone_only_emits_tombstone() {
use crate::schema::{Column, KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "ts_only_ks".to_string(),
table: "ts_only_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "name".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let pk = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let live = MergeEntry::new(
1,
pk.clone(),
None,
100,
RowData::Live {
cells: vec![CellData {
column: "name".to_string(),
value: Value::text("doomed".to_string()),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let tomb = MergeEntry::new(
0,
pk.clone(),
None,
300,
RowData::Tombstone {
deletion_time: 300,
local_deletion_time: 0,
},
);
let merger = KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let merged = merger
.merge_partition_rows(vec![tomb, live])
.expect("merge must not fail");
assert_eq!(merged.len(), 1);
match &merged[0].row_data {
RowData::Tombstone { deletion_time, .. } => {
assert_eq!(*deletion_time, 300, "tombstone deletion_time preserved");
}
other => panic!("expected a Tombstone entry, got {:?}", other),
}
}
#[test]
fn test_merge_step_variants() {
let key = DecoratedKey::new(100, vec![1, 2, 3]);
let rows = vec![];
let partition_step = MergeStep::Partition { key, rows };
match partition_step {
MergeStep::Partition { key, rows } => {
assert_eq!(key.token, 100);
assert_eq!(rows.len(), 0);
}
_ => panic!("Expected Partition variant"),
}
let complete_step = MergeStep::Complete;
match complete_step {
MergeStep::Complete => {}
_ => panic!("Expected Complete variant"),
}
}
#[test]
fn test_cell_merge_last_write_wins_higher_timestamp() {
let cell1 = CellData {
column: "name".to_string(),
value: Value::text("Old".to_string()),
timestamp: 1000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let cell2 = CellData {
column: "name".to_string(),
value: Value::text("New".to_string()),
timestamp: 2000, ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
assert!(cell2.timestamp > cell1.timestamp);
}
#[test]
fn test_memory_budget_calculation() {
let k = 10;
let buffer_size_per_run = RunReader::DEFAULT_BUFFER_SIZE;
let total_memory = k * buffer_size_per_run;
assert_eq!(buffer_size_per_run, 8 * 1024); assert_eq!(total_memory, 80 * 1024); }
#[test]
fn test_merge_entry_to_mutation_live_cells() {
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::{CellOperation, DecoratedKey};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let key_bytes = 42i32.to_be_bytes().to_vec();
let entry = MergeEntry::new(
0,
DecoratedKey::new(1000, key_bytes),
None,
999_000_000,
RowData::Live {
cells: vec![
CellData {
column: "name".to_string(),
value: Value::text("Alice".to_string()),
timestamp: 999_000_000,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
CellData {
column: "age".to_string(),
value: Value::Integer(30),
timestamp: 999_000_000,
ttl: Some(3600),
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert_eq!(mutation.partition_key.columns.len(), 1);
assert_eq!(mutation.partition_key.columns[0].0, "id");
assert_eq!(mutation.operations.len(), 2);
assert_eq!(mutation.timestamp_micros, 999_000_000);
let has_write = mutation
.operations
.iter()
.any(|op| matches!(op, CellOperation::Write { column, .. } if column == "name"));
let has_ttl_write = mutation.operations.iter().any(|op| {
matches!(op, CellOperation::WriteWithTtl { column, ttl_seconds, .. }
if column == "age" && *ttl_seconds == 3600)
});
assert!(has_write, "Expected Write operation for 'name'");
assert!(has_ttl_write, "Expected WriteWithTtl operation for 'age'");
}
#[test]
fn merge_entry_preserves_per_cell_write_timestamps() {
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::DecoratedKey;
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let cell = |column: &str, ts: i64| CellData {
column: column.to_string(),
value: Value::text(column.to_string()),
timestamp: ts,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let entry = MergeEntry::new(
0,
DecoratedKey::new(1000, 42i32.to_be_bytes().to_vec()),
None,
300,
RowData::Live {
cells: vec![cell("early", 100), cell("late", 300)],
},
);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert_eq!(mutation.timestamp_micros, 300);
let cwt = mutation
.cell_write_timestamps
.as_ref()
.expect("per-cell write timestamps must be recorded for a mixed-writetime row");
assert_eq!(cwt.get("early"), Some(&100));
assert_eq!(cwt.get("late"), None);
assert_eq!(mutation.cell_write_timestamp("early"), 100);
assert_eq!(mutation.cell_write_timestamp("late"), 300);
}
#[test]
fn merge_entry_single_writetime_row_has_no_per_cell_overrides() {
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::DecoratedKey;
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let cell = |column: &str| CellData {
column: column.to_string(),
value: Value::text(column.to_string()),
timestamp: 555,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let entry = MergeEntry::new(
0,
DecoratedKey::new(1000, 7i32.to_be_bytes().to_vec()),
None,
555,
RowData::Live {
cells: vec![cell("a"), cell("b")],
},
);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert!(
mutation.cell_write_timestamps.is_none(),
"single-writetime row must not record any per-cell overrides"
);
assert_eq!(mutation.cell_write_timestamp("a"), 555);
}
#[test]
fn test_merge_entry_to_mutation_tombstone() {
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::{CellOperation, DecoratedKey};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let key_bytes = 7i32.to_be_bytes().to_vec();
let entry = MergeEntry::new(
0,
DecoratedKey::new(500, key_bytes),
None,
888_000_000,
RowData::Tombstone {
deletion_time: 888_000_000,
local_deletion_time: 1_700_000_000,
},
);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert_eq!(mutation.operations.len(), 1);
assert!(
matches!(mutation.operations[0], CellOperation::DeleteRow),
"Expected DeleteRow operation for tombstone entry"
);
}
#[test]
fn compute_baseline_min_keeps_far_future_ldt() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let data_path = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&data_path, b"").expect("touch Data.db");
let stats_path = tmp.path().join("nb-1-big-Statistics.db");
let far_future_bits: i32 = ((1u32 << 31) + 5) as i32;
assert!(
far_future_bits < 0,
"sanity: far-future LDT is negative i32"
);
let mut meta = StatisticsMetadata::new();
meta.update_local_deletion_time(far_future_bits);
StatisticsWriter::new(stats_path)
.write(&meta, None)
.expect("write Statistics.db with far-future LDT");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[data_path]);
assert_eq!(
baseline_ldt, far_future_bits,
"far-future LDT baseline must round-trip as its i32 bit pattern, not be dropped"
);
}
#[test]
fn compute_baseline_min_skips_live_sentinel() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let data_path = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&data_path, b"").expect("touch Data.db");
let stats_path = tmp.path().join("nb-1-big-Statistics.db");
let mut meta = StatisticsMetadata::new();
meta.min_local_deletion_time = i32::MAX;
StatisticsWriter::new(stats_path)
.write(&meta, None)
.expect("write Statistics.db with LIVE sentinel");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[data_path]);
assert_eq!(
baseline_ldt,
i32::MAX,
"live/no-deletion sentinel must not lower the seeded baseline"
);
}
#[test]
fn compute_baseline_min_skips_no_deletion_sentinel_mixed_with_tombstone() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let live_data = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&live_data, b"").expect("touch live Data.db");
let live_meta = StatisticsMetadata::new(); StatisticsWriter::new(tmp.path().join("nb-1-big-Statistics.db"))
.write(&live_meta, None)
.expect("write live-only Statistics.db");
let tomb_ldt: i32 = 1_782_950_059; let tomb_data = tmp.path().join("nb-2-big-Data.db");
std::fs::write(&tomb_data, b"").expect("touch tombstone Data.db");
let mut tomb_meta = StatisticsMetadata::new();
tomb_meta.update_local_deletion_time(tomb_ldt);
StatisticsWriter::new(tmp.path().join("nb-2-big-Statistics.db"))
.write(&tomb_meta, None)
.expect("write tombstone Statistics.db");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[live_data, tomb_data]);
assert_eq!(
baseline_ldt, tomb_ldt,
"the live-only input's no-deletion sentinel must not lower the baseline \
below the tombstone input's real LDT (#1410)"
);
}
#[test]
fn compute_baseline_min_all_live_stays_unseeded() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let data_path = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&data_path, b"").expect("touch Data.db");
let meta = StatisticsMetadata::new(); StatisticsWriter::new(tmp.path().join("nb-1-big-Statistics.db"))
.write(&meta, None)
.expect("write live-only Statistics.db");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[data_path]);
assert_eq!(
baseline_ldt,
i32::MAX,
"an all-live compaction must leave the LDT baseline unseeded"
);
}
#[test]
fn compute_baseline_min_includes_genuine_zero_ldt_tombstone() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let data_path = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&data_path, b"").expect("touch Data.db");
let mut meta = StatisticsMetadata::new();
meta.update_local_deletion_time(0); StatisticsWriter::new(tmp.path().join("nb-1-big-Statistics.db"))
.write(&meta, None)
.expect("write Statistics.db with a genuine LDT-0 tombstone");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[data_path]);
assert_eq!(
baseline_ldt, 0,
"a genuine LDT-0 tombstone must be included in the baseline, not excluded \
as the no-deletion sentinel (#1410)"
);
}
#[test]
fn compute_baseline_min_conservative_when_stats_extras_unparseable() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let data_path = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&data_path, b"").expect("touch Data.db");
let stats_path = tmp.path().join("nb-1-big-Statistics.db");
let tomb_ldt: i32 = 1_782_950_059; let mut meta = StatisticsMetadata::new();
meta.update_local_deletion_time(tomb_ldt);
StatisticsWriter::new(stats_path.clone())
.write(&meta, None)
.expect("write tombstone Statistics.db");
let mut bytes = std::fs::read(&stats_path).expect("read stats");
let count = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]) as usize;
let mut stats_offset = None;
for i in 0..count {
let entry = 8 + i * 8; let ty = u32::from_be_bytes([
bytes[entry],
bytes[entry + 1],
bytes[entry + 2],
bytes[entry + 3],
]);
if ty == 2 {
stats_offset = Some(u32::from_be_bytes([
bytes[entry + 4],
bytes[entry + 5],
bytes[entry + 6],
bytes[entry + 7],
]) as usize);
}
}
let off = stats_offset.expect("STATS component (type 2) in TOC");
bytes[off..off + 4].copy_from_slice(&(-1i32).to_be_bytes());
std::fs::write(&stats_path, &bytes).expect("rewrite corrupted stats");
let (_ts, baseline_ldt, _ttl) = compute_baseline_min(&[data_path]);
assert_eq!(
baseline_ldt, tomb_ldt,
"an unparseable STATS-extras section must be treated conservatively \
(INCLUDE the LDT baseline), not as an empty no-tombstone histogram (#1410)"
);
}
#[test]
fn all_input_stats_readable_fails_closed_on_missing_or_unparseable_stats() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let good_data = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&good_data, b"").expect("touch good Data.db");
StatisticsWriter::new(tmp.path().join("nb-1-big-Statistics.db"))
.write(&StatisticsMetadata::new(), None)
.expect("write valid Statistics.db");
assert!(
all_input_stats_readable(&[good_data.clone()]),
"a well-formed input with a valid Statistics.db must be readable"
);
let missing_data = tmp.path().join("nb-2-big-Data.db");
std::fs::write(&missing_data, b"").expect("touch missing-stats Data.db");
assert!(
!all_input_stats_readable(&[missing_data.clone()]),
"an input whose Statistics.db is missing must NOT be provably deletion-free"
);
let corrupt_data = tmp.path().join("nb-3-big-Data.db");
std::fs::write(&corrupt_data, b"").expect("touch corrupt-stats Data.db");
std::fs::write(tmp.path().join("nb-3-big-Statistics.db"), b"\x00\x00")
.expect("write truncated Statistics.db");
assert!(
!all_input_stats_readable(&[corrupt_data.clone()]),
"an input whose Statistics.db is unparseable at the top level must fail closed"
);
assert!(
!all_input_stats_readable(&[good_data, missing_data, corrupt_data]),
"any single unreadable input must force the whole merge to fail closed"
);
}
#[test]
fn compute_max_purgeable_timestamp_returns_min_over_outside_sstables() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let write_stats = |gen: u32, min_ts: i64| -> PathBuf {
let data_path = tmp.path().join(format!("nb-{gen}-big-Data.db"));
std::fs::write(&data_path, b"").expect("touch Data.db");
let stats_path = tmp.path().join(format!("nb-{gen}-big-Statistics.db"));
let mut meta = StatisticsMetadata::new();
meta.min_timestamp = min_ts;
StatisticsWriter::new(stats_path)
.write(&meta, None)
.expect("write Statistics.db");
data_path
};
let a = write_stats(1, 5_000);
let b = write_stats(2, 2_500);
let c = write_stats(3, 9_000);
let bound = compute_max_purgeable_timestamp(&[a, b, c]);
assert_eq!(
bound,
Some(2_500),
"the bound must be the minimum min_timestamp across all outside SSTables"
);
assert_eq!(
compute_max_purgeable_timestamp(&[]),
None,
"no outside SSTables means no overlap bound"
);
}
#[test]
fn compute_max_purgeable_timestamp_unreadable_outside_disables_purging() {
use crate::storage::sstable::writer::{StatisticsMetadata, StatisticsWriter};
use tempfile::TempDir;
let tmp = TempDir::new().expect("temp dir");
let readable = tmp.path().join("nb-1-big-Data.db");
std::fs::write(&readable, b"").expect("touch Data.db");
let mut meta = StatisticsMetadata::new();
meta.min_timestamp = 1_000;
StatisticsWriter::new(tmp.path().join("nb-1-big-Statistics.db"))
.write(&meta, None)
.expect("write Statistics.db");
let missing = tmp.path().join("nb-2-big-Data.db");
std::fs::write(&missing, b"").expect("touch Data.db");
assert_eq!(
compute_max_purgeable_timestamp(&[readable, missing]),
None,
"an unreadable outside Statistics.db must disable overlap-aware purging"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod merge_property_tests {
use super::*;
use proptest::prelude::*;
use std::collections::HashMap;
const MERGE_TIME_SECS: i32 = 1_000;
#[derive(Debug, Clone)]
enum CellOp {
Write {
timestamp: i64,
local_deletion_time: Option<i32>,
},
Delete { timestamp: i64 },
RangeTombstone {
start_ck: u8,
end_ck: u8,
marked_for_delete_at: i64,
},
}
#[derive(Debug, Clone)]
struct CellInput {
partition: u8,
clustering: u8,
column: u8,
op: CellOp,
}
type CellKey = (u8, u8, u8);
#[derive(Debug, Clone, PartialEq, Eq)]
enum MergedCell {
Live { timestamp: i64 },
Dead { timestamp: i64 },
}
fn reference_merge(inputs: &[CellInput]) -> HashMap<CellKey, MergedCell> {
let mut per_slot: HashMap<CellKey, MergedCell> = HashMap::new();
let mut range_tombstones: Vec<CellInput> = Vec::new();
for ci in inputs {
match &ci.op {
CellOp::RangeTombstone { .. } => {
range_tombstones.push(ci.clone());
}
CellOp::Write {
timestamp,
local_deletion_time,
} => {
if local_deletion_time
.map(|ldt| ldt < MERGE_TIME_SECS)
.unwrap_or(false)
{
continue;
}
let key = (ci.partition, ci.clustering, ci.column);
let candidate = MergedCell::Live {
timestamp: *timestamp,
};
per_slot
.entry(key)
.and_modify(|existing| {
match existing {
MergedCell::Live { timestamp: ex_ts } => {
if *timestamp > *ex_ts {
*existing = candidate.clone();
}
}
MergedCell::Dead { timestamp: ex_ts } => {
if *timestamp > *ex_ts {
*existing = candidate.clone();
}
}
}
})
.or_insert(candidate);
}
CellOp::Delete { timestamp } => {
let key = (ci.partition, ci.clustering, ci.column);
let candidate = MergedCell::Dead {
timestamp: *timestamp,
};
per_slot
.entry(key)
.and_modify(|existing| {
match existing {
MergedCell::Live { timestamp: ex_ts } => {
if *timestamp >= *ex_ts {
*existing = candidate.clone();
}
}
MergedCell::Dead { timestamp: ex_ts } => {
if *timestamp > *ex_ts {
*existing = candidate.clone();
}
}
}
})
.or_insert(candidate);
}
}
}
per_slot.retain(|&(pk, ck, _col), cell| {
for rt in &range_tombstones {
if rt.partition != pk {
continue;
}
if let CellOp::RangeTombstone {
start_ck,
end_ck,
marked_for_delete_at,
} = rt.op
{
if ck >= start_ck && ck <= end_ck {
if let MergedCell::Live { timestamp } = cell {
if marked_for_delete_at >= *timestamp {
return false; }
}
}
}
}
true
});
per_slot
}
fn arb_timestamp() -> impl Strategy<Value = i64> {
1i64..=20i64
}
fn arb_local_deletion_time() -> impl Strategy<Value = Option<i32>> {
prop_oneof![
3 => Just(None), 1 => (990i32..=999i32).prop_map(Some), 1 => (1000i32..=1010i32).prop_map(Some), ]
}
fn arb_cell_op() -> impl Strategy<Value = CellOp> {
prop_oneof![
5 => (arb_timestamp(), arb_local_deletion_time())
.prop_map(|(ts, ldt)| CellOp::Write {
timestamp: ts,
local_deletion_time: ldt,
}),
3 => arb_timestamp().prop_map(|ts| CellOp::Delete { timestamp: ts }),
2 => (0u8..=3u8, 0u8..=3u8, arb_timestamp()).prop_map(|(s, e, ts)| {
let (start_ck, end_ck) = if s <= e { (s, e) } else { (e, s) };
CellOp::RangeTombstone {
start_ck,
end_ck,
marked_for_delete_at: ts,
}
}),
]
}
fn arb_cell_input() -> impl Strategy<Value = CellInput> {
(0u8..4u8, 0u8..4u8, 0u8..3u8, arb_cell_op()).prop_map(
|(partition, clustering, column, op)| CellInput {
partition,
clustering,
column,
op,
},
)
}
fn arb_cell_stream() -> impl Strategy<Value = Vec<CellInput>> {
prop::collection::vec(arb_cell_input(), 4..=32)
}
fn sorted_keys(m: &HashMap<CellKey, MergedCell>) -> Vec<(CellKey, MergedCell)> {
let mut v: Vec<_> = m.iter().map(|(&k, v)| (k, v.clone())).collect();
v.sort_by_key(|(k, _)| *k);
v
}
#[test]
fn ref_tombstone_shadows_earlier_write() {
let inputs = vec![
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Delete { timestamp: 10 },
},
];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(0, 0, 0)),
Some(&MergedCell::Dead { timestamp: 10 }),
"Delete(ts=10) must shadow Write(ts=5)"
);
}
#[test]
fn ref_write_not_shadowed_by_older_tombstone() {
let inputs = vec![
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Write {
timestamp: 10,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Delete { timestamp: 5 },
},
];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(0, 0, 0)),
Some(&MergedCell::Live { timestamp: 10 }),
"Write(ts=10) must win over Delete(ts=5)"
);
}
#[test]
fn ref_delete_wins_at_equal_timestamp() {
let inputs = vec![
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Delete { timestamp: 5 },
},
];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(0, 0, 0)),
Some(&MergedCell::Dead { timestamp: 5 }),
"Delete must win at equal timestamp (Cassandra reconcile rule)"
);
}
#[test]
fn ref_expired_ttl_drops_cell() {
let inputs = vec![CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: Some(500), },
}];
let result = reference_merge(&inputs);
assert!(
!result.contains_key(&(0, 0, 0)),
"Expired TTL cell must be absent from merged output"
);
}
#[test]
fn ref_live_ttl_keeps_cell() {
let inputs = vec![CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: Some(1500), },
}];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(0, 0, 0)),
Some(&MergedCell::Live { timestamp: 5 }),
"Non-expired TTL cell must be present"
);
}
#[test]
fn ref_range_tombstone_suppresses_row_in_range() {
let inputs = vec![
CellInput {
partition: 0,
clustering: 2,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0, column: 0,
op: CellOp::RangeTombstone {
start_ck: 0,
end_ck: 5,
marked_for_delete_at: 10,
},
},
];
let result = reference_merge(&inputs);
assert!(
!result.contains_key(&(0, 2, 0)),
"Cell with ts=5 at clustering=2 must be suppressed by RangeTombstone(mfda=10, [0,5])"
);
}
#[test]
fn ref_range_tombstone_does_not_suppress_newer_write() {
let inputs = vec![
CellInput {
partition: 0,
clustering: 2,
column: 0,
op: CellOp::Write {
timestamp: 15,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::RangeTombstone {
start_ck: 0,
end_ck: 5,
marked_for_delete_at: 10,
},
},
];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(0, 2, 0)),
Some(&MergedCell::Live { timestamp: 15 }),
"Write(ts=15) must NOT be suppressed by RangeTombstone(mfda=10)"
);
}
#[test]
fn ref_range_tombstone_only_applies_within_partition() {
let inputs = vec![
CellInput {
partition: 1,
clustering: 2,
column: 0,
op: CellOp::Write {
timestamp: 5,
local_deletion_time: None,
},
},
CellInput {
partition: 0,
clustering: 0,
column: 0,
op: CellOp::RangeTombstone {
start_ck: 0,
end_ck: 5,
marked_for_delete_at: 10,
},
},
];
let result = reference_merge(&inputs);
assert_eq!(
result.get(&(1, 2, 0)),
Some(&MergedCell::Live { timestamp: 5 }),
"RangeTombstone in partition 0 must not affect partition 1"
);
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(64))]
#[test]
fn prop_tombstone_shadowing_consistent(inputs in arb_cell_stream()) {
let merged = reference_merge(&inputs);
for (&(pk, ck, col), cell) in &merged {
if let MergedCell::Dead { timestamp: dead_ts } = cell {
let best_delete = inputs.iter()
.filter(|ci| ci.partition == pk && ci.clustering == ck && ci.column == col)
.filter_map(|ci| {
if let CellOp::Delete { timestamp } = ci.op {
Some(timestamp)
} else {
None
}
})
.max();
prop_assert!(
best_delete.is_some(),
"Dead cell at ({},{},{}) but no Delete in inputs",
pk, ck, col
);
prop_assert_eq!(
best_delete.unwrap(),
*dead_ts,
"Dead cell timestamp must equal best Delete timestamp for ({},{},{})",
pk, ck, col
);
}
}
}
#[test]
fn prop_ttl_expiry_no_expired_live_cells(inputs in arb_cell_stream()) {
let merged = reference_merge(&inputs);
for (&(pk, ck, col), cell) in &merged {
if let MergedCell::Live { .. } = cell {
let has_live_write = inputs.iter()
.filter(|ci| ci.partition == pk && ci.clustering == ck && ci.column == col)
.any(|ci| {
if let CellOp::Write { local_deletion_time, .. } = &ci.op {
local_deletion_time
.map(|ldt| ldt >= MERGE_TIME_SECS)
.unwrap_or(true)
} else {
false
}
});
prop_assert!(
has_live_write,
"Live cell at ({},{},{}) but all writes are expired",
pk, ck, col
);
}
}
}
#[test]
fn prop_range_tombstone_suppresses_covered_live_cells(inputs in arb_cell_stream()) {
let merged = reference_merge(&inputs);
let range_tombstones: Vec<(u8, u8, u8, i64)> = inputs.iter()
.filter_map(|ci| {
if let CellOp::RangeTombstone { start_ck, end_ck, marked_for_delete_at } = ci.op {
Some((ci.partition, start_ck, end_ck, marked_for_delete_at))
} else {
None
}
})
.collect();
for (&(pk, ck, _col), cell) in &merged {
if let MergedCell::Live { timestamp } = cell {
for &(rt_pk, start_ck, end_ck, mfda) in &range_tombstones {
if rt_pk == pk && ck >= start_ck && ck <= end_ck && mfda >= *timestamp {
prop_assert!(
false,
"Live cell at ({},{}) ts={} should be suppressed by \
RangeTombstone(part={}, [{},{}], mfda={})",
pk, ck, timestamp, rt_pk, start_ck, end_ck, mfda
);
}
}
}
}
}
#[test]
fn prop_live_cell_has_max_write_timestamp(inputs in arb_cell_stream()) {
let merged = reference_merge(&inputs);
for (&(pk, ck, col), cell) in &merged {
if let MergedCell::Live { timestamp: live_ts } = cell {
let max_ts = inputs.iter()
.filter(|ci| ci.partition == pk && ci.clustering == ck && ci.column == col)
.filter_map(|ci| {
if let CellOp::Write { timestamp, local_deletion_time } = &ci.op {
let not_expired = local_deletion_time
.map(|ldt| ldt >= MERGE_TIME_SECS)
.unwrap_or(true);
if not_expired { Some(*timestamp) } else { None }
} else {
None
}
})
.max();
prop_assert_eq!(
max_ts,
Some(*live_ts),
"Live cell at ({},{},{}) must have max non-expired write timestamp",
pk, ck, col
);
}
}
}
#[test]
fn prop_reference_merge_is_deterministic(inputs in arb_cell_stream()) {
let result_a = reference_merge(&inputs);
let result_b = reference_merge(&inputs);
prop_assert_eq!(
sorted_keys(&result_a),
sorted_keys(&result_b),
"reference_merge must be deterministic"
);
}
#[test]
fn prop_real_merger_lww_agrees_with_reference(
entries in prop::collection::vec(
// (clustering_key 0..4, run_index 0..2, timestamp 1..20)
(0u8..4u8, 0usize..2usize, 1i64..=20i64),
2..=12usize,
)
) {
use crate::schema::{Column, KeyColumn};
use std::collections::HashMap as SchemaMap;
let schema = TableSchema {
keyspace: "prop_test_ks".to_string(),
table: "prop_test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "value".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: SchemaMap::new(),
dropped_columns: SchemaMap::new(),
};
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let merge_entries: Vec<MergeEntry> = entries.iter().map(|&(ck, run_index, ts)| {
let ck_key = ClusteringKey {
columns: vec![("ck".to_string(), Value::TinyInt(ck as i8))],
};
MergeEntry::new(
run_index,
partition_key.clone(),
Some(ck_key),
ts,
RowData::Live {
cells: vec![CellData {
column: "value".to_string(),
value: Value::Integer(ts as i32),
timestamp: ts,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
)
}).collect();
let merger = KWayMerger {
runs: vec![],
heap: std::collections::BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema: schema.clone(),
schema_arc: std::sync::Arc::new(schema.clone()),
};
let real_merged = merger.merge_partition_rows(merge_entries.clone())
.expect("merge_partition_rows must not fail");
let mut ref_map: HashMap<u8, (i64, usize)> = HashMap::new();
for &(ck, run_index, ts) in &entries {
ref_map.entry(ck)
.and_modify(|(best_ts, best_run)| {
if ts > *best_ts || (ts == *best_ts && run_index < *best_run) {
*best_ts = ts;
*best_run = run_index;
}
})
.or_insert((ts, run_index));
}
prop_assert_eq!(
real_merged.len(),
ref_map.len(),
"real merger output row count must match reference"
);
for entry in &real_merged {
let ck_byte = match entry.clustering_key.as_ref()
.and_then(|ck| ck.columns.first())
.map(|(_, v)| v)
{
Some(Value::TinyInt(b)) => *b as u8,
_ => {
prop_assert!(false, "unexpected clustering key value");
unreachable!()
}
};
let (ref_ts, _ref_run) = ref_map[&ck_byte];
prop_assert_eq!(
entry.timestamp,
ref_ts,
"real merger winner timestamp must match reference for ck={}",
ck_byte
);
}
}
#[test]
fn prop_real_merger_tombstone_vs_live(
ts_write in 1i64..=10i64,
ts_delete in 1i64..=20i64,
) {
use crate::schema::{Column, KeyColumn};
use std::collections::HashMap as SchemaMap;
let schema = TableSchema {
keyspace: "prop_test_ks".to_string(),
table: "prop_test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "value".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: SchemaMap::new(),
dropped_columns: SchemaMap::new(),
};
let partition_key = DecoratedKey::new(100, vec![0, 0, 0, 1]);
let ck = ClusteringKey {
columns: vec![("ck".to_string(), Value::TinyInt(0))],
};
let live_entry = MergeEntry::new(
0, partition_key.clone(),
Some(ck.clone()),
ts_write,
RowData::Live {
cells: vec![CellData {
column: "value".to_string(),
value: Value::Integer(42),
timestamp: ts_write,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let tombstone_entry = MergeEntry::new(
1, partition_key.clone(),
Some(ck.clone()),
ts_delete,
RowData::Tombstone {
deletion_time: ts_delete,
local_deletion_time: 2000,
},
);
let merger = KWayMerger {
runs: vec![],
heap: std::collections::BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema: schema.clone(),
schema_arc: std::sync::Arc::new(schema.clone()),
};
let merged = merger.merge_partition_rows(vec![live_entry, tombstone_entry])
.expect("merge_partition_rows must not fail");
prop_assert_eq!(merged.len(), 1, "one clustering key => one merged row");
let winner = &merged[0];
if ts_delete > ts_write {
prop_assert!(
matches!(winner.row_data, RowData::Tombstone { .. }),
"Tombstone(ts={}) must win over Live(ts={})",
ts_delete, ts_write
);
} else if ts_write > ts_delete {
prop_assert!(
matches!(winner.row_data, RowData::Live { .. }),
"Live(ts={}) must win over Tombstone(ts={})",
ts_write, ts_delete
);
} else {
prop_assert!(
matches!(winner.row_data, RowData::Tombstone { .. }),
"At equal ts={}, Tombstone must win over Live (Cassandra reconcile rule)",
ts_delete
);
}
}
}
}
#[cfg(all(test, feature = "write-support"))]
mod streaming_tests {
use super::*;
#[test]
fn test_streaming_channel_capacity_constant() {
assert_eq!(STREAMING_CHANNEL_CAPACITY, 256);
}
struct SyntheticStreamingIterator {
rx: std::sync::mpsc::Receiver<Result<MergeEntry>>,
_tx_thread: std::thread::JoinHandle<()>,
}
impl SyntheticStreamingIterator {
fn new(count: usize, run_index: usize, capacity: usize) -> Self {
let (tx, rx) = std::sync::mpsc::sync_channel(capacity);
let tx_thread = std::thread::spawn(move || {
for i in 0..count {
let entry = MergeEntry::new(
run_index,
DecoratedKey::new(i as i64, vec![i as u8]),
None,
(i as i64) * 1000,
RowData::Live { cells: vec![] },
);
if tx.send(Ok(entry)).is_err() {
return;
}
}
});
Self {
rx,
_tx_thread: tx_thread,
}
}
}
impl SSTableRowIterator for SyntheticStreamingIterator {
fn next(&mut self) -> Option<Result<MergeEntry>> {
self.rx.recv().ok()
}
}
#[test]
fn test_kway_merge_with_streaming_sources_preserves_order() {
use crate::schema::{KeyColumn, TableSchema};
use std::collections::HashMap;
let schema = TableSchema {
keyspace: "stream_ks".to_string(),
table: "stream_tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
const N: usize = 20;
const CHANNEL_CAP: usize = 4;
let (tx0, rx0) = std::sync::mpsc::sync_channel::<Result<MergeEntry>>(CHANNEL_CAP);
let (tx1, rx1) = std::sync::mpsc::sync_channel::<Result<MergeEntry>>(CHANNEL_CAP);
std::thread::spawn(move || {
for i in 0..N {
let token = (i * 2) as i64;
let entry = MergeEntry::new(
0,
DecoratedKey::new(token, vec![(i * 2) as u8]),
None,
1000,
RowData::Live { cells: vec![] },
);
if tx0.send(Ok(entry)).is_err() {
return;
}
}
});
std::thread::spawn(move || {
for i in 0..N {
let token = (i * 2 + 1) as i64;
let entry = MergeEntry::new(
1,
DecoratedKey::new(token, vec![(i * 2 + 1) as u8]),
None,
1000,
RowData::Live { cells: vec![] },
);
if tx1.send(Ok(entry)).is_err() {
return;
}
}
});
struct ChannelIterator(std::sync::mpsc::Receiver<Result<MergeEntry>>);
impl SSTableRowIterator for ChannelIterator {
fn next(&mut self) -> Option<Result<MergeEntry>> {
self.0.recv().ok()
}
}
let runs: Vec<RunReader> = vec![
RunReader::new(Box::new(ChannelIterator(rx0))),
RunReader::new(Box::new(ChannelIterator(rx1))),
];
let mut merger = KWayMerger {
runs,
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
};
let mut token_set = std::collections::BTreeSet::new();
let mut prev_token: Option<i64> = None;
loop {
match merger.step().expect("step must not fail") {
MergeStep::Complete => break,
MergeStep::Partition { key, .. } => {
if let Some(pt) = prev_token {
assert!(
key.token >= pt,
"out-of-order token {} after {}",
key.token,
pt
);
}
prev_token = Some(key.token);
token_set.insert(key.token);
}
}
}
assert_eq!(
token_set.len(),
N * 2,
"expected {} unique partitions, got {}",
N * 2,
token_set.len()
);
for expected in 0..(N as i64 * 2) {
assert!(
token_set.contains(&expected),
"token {} is missing from merged output",
expected
);
}
}
#[test]
fn test_streaming_iterator_drains_all_entries_with_backpressure() {
const TOTAL: usize = 1000;
let mut iter = SyntheticStreamingIterator::new(TOTAL, 0, STREAMING_CHANNEL_CAPACITY);
let mut count = 0usize;
while let Some(result) = iter.next() {
result.expect("entry must not be an error");
count += 1;
}
assert_eq!(count, TOTAL, "all {} entries must be produced", TOTAL);
}
#[test]
fn test_run_reader_with_streaming_source() {
const N: usize = 50;
let iter = SyntheticStreamingIterator::new(N, 0, 4);
let mut reader = RunReader::new(Box::new(iter));
let mut seen = 0usize;
loop {
match reader.peek().expect("peek must not error") {
None => break,
Some(_) => {
reader.advance().expect("advance must not error");
seen += 1;
}
}
}
assert_eq!(seen, N, "RunReader must surface all {} entries", N);
assert!(
reader.is_exhausted(),
"RunReader must be exhausted after drain"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_823_complex_column_merge {
use super::*;
use crate::types::{TombstoneInfo, TombstoneType, UdtField, UdtValue, Value};
fn dk(byte: u8) -> DecoratedKey {
DecoratedKey::from_key_bytes(vec![byte]).expect("token")
}
fn live(run_index: usize, row_ts: i64, cells: Vec<CellData>) -> MergeEntry {
MergeEntry::new(run_index, dk(1), None, row_ts, RowData::Live { cells })
}
fn scalar_cell(column: &str, value: &str, ts: i64) -> CellData {
CellData::new(column.to_string(), Value::text(value.to_string()), ts)
}
#[test]
fn dropped_column_cell_at_or_before_drop_time_is_filtered() {
let row = live(
0,
200,
vec![
scalar_cell("name", "alice", 100),
scalar_cell("legacy", "stale", 100),
],
);
let mut dropped = ::std::collections::HashMap::new();
dropped.insert("legacy".to_string(), 150);
let merged = KWayMerger::reconcile_cluster(None, vec![row], &dropped, None)
.expect("a live row must be emitted (name survives)");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1, "the dropped-column cell must be discarded");
assert_eq!(cells[0].column, "name");
}
#[test]
fn dropped_column_cell_after_drop_time_survives() {
let row = live(0, 200, vec![scalar_cell("legacy", "fresh", 200)]);
let mut dropped = ::std::collections::HashMap::new();
dropped.insert("legacy".to_string(), 150);
let merged = KWayMerger::reconcile_cluster(None, vec![row], &dropped, None)
.expect("a live row must be emitted (cell post-dates the drop)");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1);
assert_eq!(cells[0].column, "legacy");
assert_eq!(cells[0].value, Value::text("fresh".to_string()));
}
#[test]
fn dropped_column_cell_at_exact_drop_time_is_filtered() {
let row = live(0, 150, vec![scalar_cell("legacy", "edge", 150)]);
let mut dropped = ::std::collections::HashMap::new();
dropped.insert("legacy".to_string(), 150);
assert!(
KWayMerger::reconcile_cluster(None, vec![row], &dropped, None).is_none(),
"cell at exactly drop_time must be discarded, leaving no surviving cells"
);
}
#[test]
fn all_cells_dropped_yields_no_row() {
let row = live(
0,
120,
vec![scalar_cell("a", "x", 100), scalar_cell("b", "y", 110)],
);
let mut dropped = ::std::collections::HashMap::new();
dropped.insert("a".to_string(), 200);
dropped.insert("b".to_string(), 200);
assert!(
KWayMerger::reconcile_cluster(None, vec![row], &dropped, None).is_none(),
"a row whose every cell is a dropped-column cell emits nothing"
);
}
#[test]
fn empty_dropped_map_is_noop() {
let row = live(
0,
200,
vec![
scalar_cell("name", "alice", 100),
scalar_cell("legacy", "stale", 100),
],
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![row],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 2, "no drops configured → every cell survives");
}
#[test]
fn dropped_column_purge_is_exact_per_cell_within_one_row() {
let row = live(
0,
300, vec![
scalar_cell("name", "alice", 300),
scalar_cell("legacy", "stale", 100),
],
);
let mut dropped = ::std::collections::HashMap::new();
dropped.insert("legacy".to_string(), 150);
let merged = KWayMerger::reconcile_cluster(None, vec![row], &dropped, None)
.expect("name survives, so a live row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1, "only the post-drop `name` cell survives");
assert_eq!(cells[0].column, "name");
assert_eq!(cells[0].timestamp, 300);
}
#[test]
fn multicell_collection_collapses_whole_column_not_per_path() {
let newer = live(
0,
200,
vec![CellData {
column: "tags".to_string(),
value: Value::List(vec![Value::text("b".to_string())]),
timestamp: 200,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
);
let older = live(
1,
100,
vec![CellData {
column: "tags".to_string(),
value: Value::List(vec![Value::text("a".to_string())]),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![newer, older],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(
cells.len(),
1,
"column-name keyed merge collapses to one cell"
);
assert_eq!(cells[0].column, "tags");
assert_eq!(
cells[0].value,
Value::List(vec![Value::text("b".to_string())]),
"winner is the higher-timestamp WHOLE collection value, not a union \
of [a, b] — confirms whole-group collapse, NOT per-path merge (#18)"
);
}
#[test]
fn nonfrozen_udt_collapses_whole_column_not_per_field() {
let mk_udt = |field: &str, v: &str| {
Value::Udt(Box::new(UdtValue {
type_name: "addr".to_string(),
keyspace: "ks".to_string(),
fields: vec![UdtField {
name: field.to_string(),
value: Some(Value::text(v.to_string())),
}],
}))
};
let newer = live(
0,
200,
vec![CellData {
column: "address".to_string(),
value: mk_udt("city", "SF"),
timestamp: 200,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
);
let older = live(
1,
100,
vec![CellData {
column: "address".to_string(),
value: mk_udt("zip", "94105"),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![newer, older],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(
cells.len(),
1,
"UDT column collapses to one whole-value cell"
);
assert_eq!(
cells[0].value,
mk_udt("city", "SF"),
"newer whole-UDT value wins; older field write is dropped — no per-field \
merge, so #18 path-ordering does not apply"
);
}
#[test]
fn adapter_produces_one_cell_per_top_level_column() {
let row = Value::Map(vec![
(Value::text("id".to_string()), Value::text("k1".to_string())),
(
Value::text("tags".to_string()),
Value::List(vec![
Value::text("a".to_string()),
Value::text("b".to_string()),
]),
),
]);
let row_data = SSTableRowIteratorAdapter::value_to_row_data(&row, 500).expect("row data");
let cells = match row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 2, "one CellData per top-level column");
let tags = cells
.iter()
.find(|c| c.column == "tags")
.expect("tags cell present");
assert_eq!(
tags.value,
Value::List(vec![
Value::text("a".to_string()),
Value::text("b".to_string())
]),
"the collection arrives as ONE nested Value, never as per-path cells"
);
}
#[test]
fn row_deletion_supersedes_equal_ts_cell_tombstone() {
let row_tomb = MergeEntry::new(
0,
dk(1),
None,
100,
RowData::Tombstone {
deletion_time: 100,
local_deletion_time: 0,
},
);
let cell_tomb = live(
1,
100,
vec![CellData {
column: "tags".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: 100,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: 0,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: 100,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![row_tomb, cell_tomb],
&::std::collections::HashMap::new(),
None,
)
.expect("row tombstone keeps the row shadowed");
match merged.row_data {
RowData::Tombstone { deletion_time, .. } => {
assert_eq!(deletion_time, 100, "row tombstone preserved at its ts");
}
RowData::Live { cells } => {
panic!("expected row to stay shadowed, got live cells: {:?}", cells)
}
}
}
fn expiring_cell(value: &str, ts: i64, ttl: u32, ldt: i32) -> CellData {
CellData {
column: "v".to_string(),
value: Value::text(value.to_string()),
timestamp: ts,
ttl: Some(ttl),
cell_path: None,
local_deletion_time: Some(ldt),
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}
}
fn cell_tombstone(ts: i64, ldt: i32) -> CellData {
CellData {
column: "v".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: ts,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: ldt as i64,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: ts,
ttl: None,
cell_path: None,
local_deletion_time: Some(ldt),
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}
}
#[test]
fn issue_848_tombstone_beats_expiring_at_equal_ts_expiring_first() {
const TS: i64 = 200;
let expiring = live(
0,
TS,
vec![expiring_cell("resurrected-if-buggy", TS, 3600, 9_999)],
);
let tombstone = live(1, TS, vec![cell_tombstone(TS, 1_000)]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![expiring, tombstone],
&::std::collections::HashMap::new(),
None,
)
.expect("a row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1, "single column `v`");
assert!(
KWayMerger::is_cell_tombstone(&cells[0]),
"at equal ts the cell TOMBSTONE must win over the expiring cell \
(before any localDeletionTime compare); got {:?}",
cells[0].value
);
assert!(
cells[0].ttl.is_none(),
"the surviving winner is the tombstone, which carries no TTL"
);
}
#[test]
fn issue_848_tombstone_beats_expiring_at_equal_ts_tombstone_first() {
const TS: i64 = 200;
let tombstone = live(0, TS, vec![cell_tombstone(TS, 1_000)]);
let expiring = live(
1,
TS,
vec![expiring_cell("resurrected-if-buggy", TS, 3600, 9_999)],
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![tombstone, expiring],
&::std::collections::HashMap::new(),
None,
)
.expect("a row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1, "single column `v`");
assert!(
KWayMerger::is_cell_tombstone(&cells[0]),
"at equal ts the cell TOMBSTONE must win regardless of source order; \
got {:?}",
cells[0].value
);
}
#[test]
fn issue_848_newer_expiring_beats_older_tombstone() {
let tombstone = live(0, 100, vec![cell_tombstone(100, 1_000)]);
let expiring = live(1, 200, vec![expiring_cell("survives", 200, 3600, 9_999)]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![tombstone, expiring],
&::std::collections::HashMap::new(),
None,
)
.expect("a row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {:?}", other),
};
assert_eq!(cells.len(), 1, "single column `v`");
assert!(
!KWayMerger::is_cell_tombstone(&cells[0]),
"a strictly NEWER expiring cell (ts=200) beats the older tombstone \
(ts=100); the deletion tie-break only fires at equal ts"
);
assert_eq!(cells[0].value, Value::text("survives".to_string()));
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_886_merge_entry_enrichment {
use super::*;
use crate::storage::write_engine::mutation::ClusteringBound;
use crate::types::Value;
fn dk(byte: u8) -> DecoratedKey {
DecoratedKey::from_key_bytes(vec![byte]).expect("token")
}
#[test]
fn celldata_new_defaults_enriched_fields_to_none() {
let cell = CellData::new("c".to_string(), Value::Integer(7), 100);
assert_eq!(cell.ttl, None);
assert_eq!(
cell.local_deletion_time, None,
"LDT defaults None (plumbing)"
);
assert_eq!(cell.cell_path, None, "cell_path defaults None (plumbing)");
}
#[test]
fn enriched_celldata_round_trips_through_merge_entry() {
let cell = CellData {
column: "m".to_string(),
value: Value::text("v".to_string()),
timestamp: 500,
ttl: Some(3600),
local_deletion_time: Some(1_700_000_000),
cell_path: Some(vec![0x00, 0x01]),
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let entry = MergeEntry::new(
0,
dk(1),
None,
500,
RowData::Live {
cells: vec![cell.clone()],
},
);
let cloned = entry.clone();
let cells = match cloned.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(cells.len(), 1);
assert_eq!(cells[0].local_deletion_time, Some(1_700_000_000));
assert_eq!(cells[0].cell_path, Some(vec![0x00, 0x01]));
assert_eq!(cells[0].ttl, Some(3600));
assert_eq!(cells[0], cell, "enriched cell survives clone + equality");
}
#[test]
fn value_to_row_data_threads_enriched_fields_as_none() {
let map = Value::Map(vec![(
Value::text("name".to_string()),
Value::text("alice".to_string()),
)]);
let row_data = SSTableRowIteratorAdapter::value_to_row_data(&map, 100)
.expect("value_to_row_data must succeed");
let cells = match row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(cells.len(), 1);
assert_eq!(cells[0].column, "name");
assert_eq!(cells[0].timestamp, 100, "live cell inherits row ts (#533)");
assert_eq!(cells[0].local_deletion_time, None);
assert_eq!(cells[0].cell_path, None);
assert_eq!(cells[0].ttl, None);
let single = SSTableRowIteratorAdapter::value_to_row_data(&Value::Integer(42), 200)
.expect("value_to_row_data must succeed");
let cells = match single {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(cells[0].local_deletion_time, None);
assert_eq!(cells[0].cell_path, None);
}
#[test]
fn merge_entry_carries_complex_deletion_without_acting_on_it() {
let complex = ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 1234,
local_deletion_time: 1_700_000_000,
};
let entry = MergeEntry::new(0, dk(1), None, 100, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![complex.clone()]);
assert_eq!(entry.complex_deletions, vec![complex]);
let cell = CellData::new("tags".to_string(), Value::text("a".to_string()), 1234);
let live = MergeEntry::new(0, dk(1), None, 1234, RowData::Live { cells: vec![cell] })
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 1234,
local_deletion_time: 1_700_000_000,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![live],
&::std::collections::HashMap::new(),
None,
)
.expect("live row must be emitted");
match merged.row_data {
RowData::Live { cells } => {
assert_eq!(
cells.len(),
1,
"complex deletion must NOT shadow (plumbing)"
);
assert_eq!(cells[0].column, "tags");
}
other => panic!("expected Live, got {other:?}"),
}
}
#[test]
fn merge_entry_carries_range_deletion_without_acting_on_it() {
let range = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 5000,
local_deletion_time: 0,
};
let cell = CellData::new("v".to_string(), Value::Integer(1), 1000);
let entry = MergeEntry::new(0, dk(1), None, 1000, RowData::Live { cells: vec![cell] })
.with_range_deletion(range.clone());
assert_eq!(entry.range_deletion, Some(range));
let merged = KWayMerger::reconcile_cluster(
None,
vec![entry],
&::std::collections::HashMap::new(),
None,
)
.expect("live row must be emitted");
match merged.row_data {
RowData::Live { cells } => {
assert_eq!(
cells.len(),
1,
"range deletion must NOT shadow covered cell (plumbing)"
);
}
other => panic!("expected Live, got {other:?}"),
}
}
#[test]
fn reconcile_cluster_preserves_carried_deletion_metadata() {
let complex_a = ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 1234,
local_deletion_time: 1_700_000_000,
};
let complex_b = ComplexDeletion {
column: "notes".to_string(),
marked_for_delete_at: 999,
local_deletion_time: 1_700_000_001,
};
let range_low = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 3000,
local_deletion_time: 0,
};
let range_high = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 7000,
local_deletion_time: 0,
};
let row0 = MergeEntry::new(
0,
dk(1),
None,
2000,
RowData::Live {
cells: vec![CellData::new("v".to_string(), Value::Integer(2), 2000)],
},
)
.with_complex_deletions(vec![complex_a.clone()])
.with_range_deletion(range_low.clone());
let row1 = MergeEntry::new(
1,
dk(1),
None,
1000,
RowData::Live {
cells: vec![CellData::new("w".to_string(), Value::Integer(1), 1000)],
},
)
.with_complex_deletions(vec![complex_a.clone(), complex_b.clone()])
.with_range_deletion(range_high.clone());
let merged = KWayMerger::reconcile_cluster(
None,
vec![row0, row1],
&::std::collections::HashMap::new(),
None,
)
.expect("live row must be emitted");
assert_eq!(
merged.complex_deletions,
vec![complex_a.clone(), complex_b.clone()],
"complex deletions: first-seen union, deduplicated (Phase A neutral)"
);
assert_eq!(
merged.range_deletion,
Some(range_high),
"range deletion with the highest deletion timestamp must be carried"
);
let plain0 = MergeEntry::new(
0,
dk(1),
None,
2000,
RowData::Live {
cells: vec![CellData::new("v".to_string(), Value::Integer(2), 2000)],
},
);
let plain1 = MergeEntry::new(
1,
dk(1),
None,
1000,
RowData::Live {
cells: vec![CellData::new("w".to_string(), Value::Integer(1), 1000)],
},
);
let plain = KWayMerger::reconcile_cluster(
None,
vec![plain0, plain1],
&::std::collections::HashMap::new(),
None,
)
.expect("live row must be emitted");
assert_eq!(
merged.row_data, plain.row_data,
"carrying deletion metadata must not change surviving-cell output"
);
assert_eq!(merged.timestamp, plain.timestamp);
assert!(plain.complex_deletions.is_empty());
assert_eq!(plain.range_deletion, None);
}
#[test]
fn reconcile_cluster_preserves_metadata_on_tombstone_entry() {
let complex = ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 10,
local_deletion_time: 1_700_000_000,
};
let row = MergeEntry::new(
0,
dk(1),
None,
500,
RowData::Tombstone {
deletion_time: 500,
local_deletion_time: 0,
},
)
.with_complex_deletions(vec![complex.clone()]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![row],
&::std::collections::HashMap::new(),
None,
)
.expect("row tombstone must be emitted");
assert!(matches!(merged.row_data, RowData::Tombstone { .. }));
assert_eq!(merged.complex_deletions, vec![complex]);
}
#[test]
fn reconcile_cluster_emits_metadata_only_entry_when_no_row_produced() {
let complex = ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 4242,
local_deletion_time: 1_700_000_000,
};
let range = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 8888,
local_deletion_time: 0,
};
let row = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![complex.clone()])
.with_range_deletion(range.clone());
let merged = KWayMerger::reconcile_cluster(
None,
vec![row],
&::std::collections::HashMap::new(),
None,
)
.expect("metadata-only cluster must still emit an entry");
match &merged.row_data {
RowData::Live { cells } => {
assert!(
cells.is_empty(),
"metadata-only entry must have no live cells"
);
}
other => panic!("expected empty Live, got {other:?}"),
}
assert_eq!(merged.complex_deletions, vec![complex]);
assert_eq!(merged.range_deletion, Some(range));
}
#[test]
fn reconcile_cluster_empty_live_without_metadata_yields_none() {
let row = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] });
assert!(
KWayMerger::reconcile_cluster(
None,
vec![row],
&::std::collections::HashMap::new(),
None
)
.is_none(),
"empty live row with no metadata must not emit an entry"
);
}
#[test]
fn complex_deletion_only_entry_reaches_writer_but_range_only_is_skipped() {
let complex = ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 4242,
local_deletion_time: 1_700_000_000,
};
let range = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 8888,
local_deletion_time: 0,
};
let complex_only = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![complex.clone()]);
assert!(
!complex_only.is_metadata_only_no_op(),
"Phase C: a complex-deletion-only entry must reach the writer"
);
let range_only = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_range_deletion(range.clone());
assert!(
!range_only.is_metadata_only_no_op(),
"#933: a range-deletion-only carrier must reach the writer"
);
let both = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![complex])
.with_range_deletion(range);
assert!(
!both.is_metadata_only_no_op(),
"a carried complex deletion keeps the entry writer-visible"
);
let reconciled = KWayMerger::reconcile_cluster(
None,
vec![complex_only],
&::std::collections::HashMap::new(),
None,
)
.expect("complex-deletion-only cluster must still emit an entry");
assert!(
!reconciled.is_metadata_only_no_op(),
"the reconciled complex-deletion entry must reach the writer"
);
}
#[test]
fn non_metadata_only_entries_are_not_skipped() {
let live = MergeEntry::new(
0,
dk(1),
None,
100,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("v".to_string()),
100,
)],
},
);
assert!(!live.is_metadata_only_no_op());
let empty_no_meta = MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] });
assert!(empty_no_meta.is_metadata_only_no_op());
let tomb = MergeEntry::new(
0,
dk(1),
None,
500,
RowData::Tombstone {
deletion_time: 500,
local_deletion_time: 0,
},
)
.with_range_deletion(RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time: 8888,
local_deletion_time: 0,
});
assert!(
!tomb.is_metadata_only_no_op(),
"a row tombstone is real content even when carrying range metadata"
);
}
#[test]
fn default_carried_fields_preserve_merge_entry_equality() {
let a = MergeEntry::new(0, dk(1), None, 100, RowData::Live { cells: vec![] });
let b = MergeEntry::new(0, dk(1), None, 100, RowData::Live { cells: vec![] });
assert_eq!(a, b);
assert!(a.complex_deletions.is_empty());
assert_eq!(a.range_deletion, None);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_899_per_element_merge {
use super::*;
use crate::storage::sstable::reader::compaction_row::{
CompactionRowData, ComplexColumn, ComplexElement,
};
use crate::types::Value;
fn dk(byte: u8) -> DecoratedKey {
DecoratedKey::from_key_bytes(vec![byte]).expect("token")
}
fn element(path: &[u8], val: &str, ts: i64) -> ComplexElement {
ComplexElement {
cell_path: path.to_vec(),
value: Some(Value::text(val.to_string())),
decoded_key: None,
timestamp: ts,
ttl: None,
local_deletion_time: None,
is_deleted: false,
has_empty_value: false,
}
}
#[test]
fn multi_cell_collection_surfaces_one_celldata_per_element() {
let elements = vec![
element(&[0xAA], "a", 100),
element(&[0xBB], "b", 200),
element(&[0xCC], "c", 300),
];
assert_eq!(elements.len(), 3);
let row_data = CompactionRowData::Live {
simple: vec![],
complex: vec![ComplexColumn {
column: "tags".to_string(),
complex_deletion: None,
elements,
collapsed_value: Value::Set(vec![
Value::text("a".to_string()),
Value::text("b".to_string()),
Value::text("c".to_string()),
]),
}],
row_deletion: None,
};
let (row, complex_deletions, _row_deletion) =
SSTableRowIteratorAdapter::compaction_row_data_to_row_data(row_data, 999);
assert!(complex_deletions.is_empty(), "no complex deletion present");
let cells = match row {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(cells.len(), 3, "one CellData per element (Phase C flip)");
for c in &cells {
assert_eq!(c.column, "tags");
assert!(c.is_complex_element, "each is a complex element");
assert!(!c.is_deleted, "live elements");
assert!(c.cell_path.is_some(), "per-element cell_path populated");
}
let mut by_path: Vec<(Vec<u8>, i64, Value)> = cells
.into_iter()
.map(|c| (c.cell_path.expect("path"), c.timestamp, c.value))
.collect();
by_path.sort_by(|a, b| a.0.cmp(&b.0));
assert_eq!(by_path[0], (vec![0xAA], 100, Value::text("a".to_string())));
assert_eq!(by_path[1], (vec![0xBB], 200, Value::text("b".to_string())));
assert_eq!(by_path[2], (vec![0xCC], 300, Value::text("c".to_string())));
}
#[test]
fn complex_deletion_reaches_merge_entry_complex_deletions() {
let row_data = CompactionRowData::Live {
simple: vec![],
complex: vec![ComplexColumn {
column: "tags".to_string(),
complex_deletion: Some((12_345, 1_700_000_000)),
elements: vec![element(&[0xAB], "x", 20_000)],
collapsed_value: Value::Set(vec![Value::text("x".to_string())]),
}],
row_deletion: None,
};
let (_row, complex_deletions, _row_deletion) =
SSTableRowIteratorAdapter::compaction_row_data_to_row_data(row_data, 20_000);
assert_eq!(complex_deletions.len(), 1);
assert_eq!(complex_deletions[0].column, "tags");
assert_eq!(complex_deletions[0].marked_for_delete_at, 12_345);
assert_eq!(complex_deletions[0].local_deletion_time, 1_700_000_000);
}
#[test]
fn reconcile_keeps_disjoint_elements_per_cell_path() {
let newer = MergeEntry::new(
0,
dk(1),
None,
200,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("b".to_string()),
timestamp: 200,
ttl: None,
cell_path: Some(vec![0xBB]),
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let older = MergeEntry::new(
1,
dk(1),
None,
100,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("a".to_string()),
timestamp: 100,
ttl: None,
cell_path: Some(vec![0xAA]),
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
);
let merged = KWayMerger::reconcile_cluster(
None,
vec![newer, older],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(cells.len(), 2, "disjoint elements both survive");
let mut paths: Vec<Vec<u8>> = cells
.iter()
.map(|c| c.cell_path.clone().expect("cell_path"))
.collect();
paths.sort();
assert_eq!(paths, vec![vec![0xAA], vec![0xBB]]);
}
#[test]
fn complex_deletion_shadows_covered_elements_before_purge() {
let old_el = MergeEntry::new(
1,
dk(1),
None,
100,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("old".to_string()),
timestamp: 100,
ttl: None,
cell_path: Some(vec![0x01]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
}],
},
);
let new_el = MergeEntry::new(
0,
dk(1),
None,
300,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("new".to_string()),
timestamp: 300,
ttl: None,
cell_path: Some(vec![0x02]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
}],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 200,
local_deletion_time: 1_700_000_000,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![new_el, old_el],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
assert_eq!(
merged.complex_deletions,
vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 200,
local_deletion_time: 1_700_000_000,
}],
"complex deletion is carried on the MergeEntry"
);
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(
cells.len(),
1,
"the covered element (ts<=mfda) is shadowed before purge (#887 f66fa14f)"
);
assert_eq!(
cells[0].cell_path.as_deref(),
Some([0x02].as_slice()),
"the strictly-newer element (ts>mfda) survives"
);
}
#[test]
fn complex_deletion_equal_timestamps_do_not_supersede() {
let covered = MergeEntry::new(
1,
dk(1),
None,
200,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("at-mfda".to_string()),
timestamp: 200,
ttl: None,
cell_path: Some(vec![0x01]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
}],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 200,
local_deletion_time: 1_700_000_100,
}]);
let survivor = MergeEntry::new(
0,
dk(1),
None,
300,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("after".to_string()),
timestamp: 300,
ttl: None,
cell_path: Some(vec![0x02]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
}],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 200,
local_deletion_time: 1_700_000_000,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![survivor, covered],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let tags_dels: Vec<&ComplexDeletion> = merged
.complex_deletions
.iter()
.filter(|d| d.column == "tags")
.collect();
assert_eq!(
tags_dels.len(),
1,
"the union keeps one complex deletion for the column"
);
assert_eq!(
tags_dels[0].marked_for_delete_at, 200,
"equal-ts deletions do not produce a stronger mfda (bd244649)"
);
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(
cells.len(),
1,
"only the strictly-newer element survives the active deletion"
);
assert_eq!(
cells[0].cell_path.as_deref(),
Some([0x02].as_slice()),
"ts>mfda element survives; equal-ts deletions did not shadow it"
);
}
#[test]
fn complex_deletion_strict_supersede_shadows_before_purge() {
let weak = MergeEntry::new(
1,
dk(1),
None,
500,
RowData::Live {
cells: vec![
CellData {
column: "tags".to_string(),
value: Value::text("ancient".to_string()),
timestamp: 50,
ttl: None,
cell_path: Some(vec![0x01]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
},
CellData {
column: "tags".to_string(),
value: Value::text("survivor".to_string()),
timestamp: 500,
ttl: None,
cell_path: Some(vec![0x03]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
},
],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 100,
local_deletion_time: 1_700_000_000,
}]);
let strong = MergeEntry::new(
0,
dk(1),
None,
300,
RowData::Live {
cells: vec![CellData {
column: "tags".to_string(),
value: Value::text("covered".to_string()),
timestamp: 300,
ttl: None,
cell_path: Some(vec![0x02]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
}],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 300,
local_deletion_time: 1_700_000_200,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![strong, weak],
&::std::collections::HashMap::new(),
None,
)
.expect("a live row must be emitted");
let tags_dels: Vec<&ComplexDeletion> = merged
.complex_deletions
.iter()
.filter(|d| d.column == "tags")
.collect();
assert_eq!(
tags_dels.len(),
1,
"one surviving complex deletion for tags"
);
assert_eq!(
tags_dels[0].marked_for_delete_at, 300,
"the strictly-greater mfda supersedes (bd244649)"
);
let cells = match merged.row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(
cells.len(),
1,
"all elements with ts<=mfda are shadowed before purge (no resurrection)"
);
assert_eq!(
cells[0].cell_path.as_deref(),
Some([0x03].as_slice()),
"the strictly-newer element (ts>mfda) survives"
);
assert_eq!(cells[0].timestamp, 500);
}
#[test]
fn row_tombstone_keeps_strictly_newer_complex_deletion_marker() {
use crate::schema::{Column, KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::CellOperation;
use std::collections::HashMap;
const T_LOW: i64 = 100; const T_HIGH: i64 = 300;
let row_tomb = MergeEntry::new(
0,
dk(1),
None,
T_LOW,
RowData::Tombstone {
deletion_time: T_LOW,
local_deletion_time: 0,
},
);
let carrier = MergeEntry::new(
1,
dk(1),
None,
T_HIGH + 200,
RowData::Live {
cells: vec![
CellData {
column: "tags".to_string(),
value: Value::text("shadowed_by_row".to_string()),
timestamp: T_LOW,
ttl: None,
cell_path: Some(vec![0x01]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
},
CellData {
column: "tags".to_string(),
value: Value::text("shadowed_by_complex".to_string()),
timestamp: T_HIGH,
ttl: None,
cell_path: Some(vec![0x02]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
},
CellData {
column: "tags".to_string(),
value: Value::text("survivor".to_string()),
timestamp: T_HIGH + 200,
ttl: None,
cell_path: Some(vec![0x03]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
},
],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: T_HIGH,
local_deletion_time: 1_700_000_000,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![row_tomb, carrier],
&::std::collections::HashMap::new(),
None,
)
.expect("a row must be emitted (it carries a row tombstone + survivor)");
let cells = match &merged.row_data {
RowData::Live { cells } => cells.clone(),
other => panic!("expected Live (survivor present), got {other:?}"),
};
assert_eq!(
cells.len(),
1,
"only the element with ts > T_high survives the complex deletion"
);
assert_eq!(cells[0].cell_path.as_deref(), Some([0x03].as_slice()));
assert_eq!(cells[0].timestamp, T_HIGH + 200);
let tags_dels: Vec<&ComplexDeletion> = merged
.complex_deletions
.iter()
.filter(|d| d.column == "tags")
.collect();
assert_eq!(tags_dels.len(), 1, "the T_high complex deletion is carried");
assert_eq!(tags_dels[0].marked_for_delete_at, T_HIGH);
let schema = TableSchema {
keyspace: "ks".to_string(),
table: "t".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "tags".to_string(),
data_type: "list<text>".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let int_pk = DecoratedKey::new(1, 1i32.to_be_bytes().to_vec());
let tomb_entry = MergeEntry::new(
0,
int_pk,
None,
T_LOW,
RowData::Tombstone {
deletion_time: T_LOW,
local_deletion_time: 0,
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: T_HIGH,
local_deletion_time: 1_700_000_000,
}]);
let mutation = KWayMerger::merge_entry_to_mutation(tomb_entry, &schema)
.expect("conversion should succeed");
let has_delete_row = mutation
.operations
.iter()
.any(|op| matches!(op, CellOperation::DeleteRow));
assert!(has_delete_row, "the row tombstone must still be emitted");
let strictly_newer_marker = mutation.operations.iter().any(|op| {
matches!(
op,
CellOperation::ComplexDeletion {
column,
marked_for_delete_at,
..
} if column == "tags" && *marked_for_delete_at == T_HIGH
)
});
assert!(
strictly_newer_marker,
"the strictly-newer (mfda > row_del) complex deletion marker must be \
emitted alongside the row tombstone (else (row_del, mfda] resurrects)"
);
}
#[test]
fn row_tombstone_drops_fully_covered_complex_deletion_marker() {
use crate::schema::{Column, KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::CellOperation;
use std::collections::HashMap;
const ROW_DEL: i64 = 300;
let schema = TableSchema {
keyspace: "ks".to_string(),
table: "t".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![Column {
name: "tags".to_string(),
data_type: "list<text>".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let int_pk = DecoratedKey::new(1, 1i32.to_be_bytes().to_vec());
for covered_mfda in [ROW_DEL - 100, ROW_DEL] {
let entry = MergeEntry::new(
0,
int_pk.clone(),
None,
ROW_DEL,
RowData::Tombstone {
deletion_time: ROW_DEL,
local_deletion_time: 0,
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: covered_mfda,
local_deletion_time: 1_700_000_000,
}]);
let mutation = KWayMerger::merge_entry_to_mutation(entry, &schema)
.expect("conversion should succeed");
assert!(
mutation
.operations
.iter()
.all(|op| !matches!(op, CellOperation::ComplexDeletion { .. })),
"a marker with mfda ({covered_mfda}) <= row_del ({ROW_DEL}) is fully \
covered by the row tombstone and must be dropped (strict boundary)"
);
assert!(
matches!(mutation.operations.as_slice(), [CellOperation::DeleteRow]),
"the fully-covered case reduces to exactly [DeleteRow]"
);
}
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_822_merge_ordering_semantics {
use super::*;
use crate::schema::{ClusteringColumn, ClusteringOrder, Column, KeyColumn, TableSchema};
use crate::storage::sstable::writer::{DataWriter, StatisticsMetadata};
use crate::storage::write_engine::mutation::{
CellOperation, ClusteringKey, DecoratedKey, Mutation, PartitionKey, TableId,
};
use crate::types::{TombstoneInfo, TombstoneType, Value};
use std::collections::HashMap;
const ROW_HAS_TIMESTAMP: u8 = 0x04;
const ROW_HAS_DELETION: u8 = 0x10;
const ROW_HAS_ALL_COLUMNS: u8 = 0x20;
const ROW_HAS_EXTENDED_FLAGS: u8 = 0x80;
const EXTENDED_IS_STATIC: u8 = 0x01;
const CELL_IS_DELETED: u8 = 0x01;
const CELL_IS_EXPIRING: u8 = 0x02;
fn writer_stats() -> StatisticsMetadata {
let mut s = StatisticsMetadata::new();
s.min_timestamp = 1_000_000;
s.min_ttl = 0;
s.min_local_deletion_time = 0;
s
}
fn read_vuint(data: &[u8], pos: usize) -> (u64, usize) {
let first = data[pos];
let extra = first.leading_ones() as usize;
assert!(extra < 8, "9-byte vint not expected in this framing");
let mask: u64 = 0xFFu64 >> (extra + 1);
let mut value = (first as u64) & mask;
for i in 0..extra {
value = (value << 8) | data[pos + 1 + i] as u64;
}
(value, extra + 1)
}
fn int_key_bytes(n: i32) -> Vec<u8> {
n.to_be_bytes().to_vec()
}
const INT_PK_HEADER_SIZE: usize = 2 + 4 + 4 + 8;
fn schema_one_clustering(ck_name: &str, ck_type: &str, order: ClusteringOrder) -> TableSchema {
TableSchema {
keyspace: "issue_822".to_string(),
table: "tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: ck_name.to_string(),
data_type: ck_type.to_string(),
position: 0,
order,
}],
columns: vec![Column {
name: "v".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn empty_merger(schema: TableSchema) -> KWayMerger {
KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
}
}
fn ck_text(col: &str, s: &str) -> ClusteringKey {
ClusteringKey {
columns: vec![(col.to_string(), Value::text(s.to_string()))],
}
}
#[test]
fn issue_10_desc_empty_vs_valued_via_clustering_compare() {
let empty = ck_text("ck", "");
let valued = ck_text("ck", "a");
let asc = schema_one_clustering("ck", "text", ClusteringOrder::Asc);
assert_eq!(
empty.compare(&valued, &asc).expect("compare must not fail"),
Ordering::Less,
"ASC: empty clustering value must sort BEFORE a valued one"
);
let desc = schema_one_clustering("ck", "text", ClusteringOrder::Desc);
assert_eq!(
empty
.compare(&valued, &desc)
.expect("compare must not fail"),
Ordering::Greater,
"DESC: empty clustering value must sort AFTER a valued one (reversed-ness \
must apply to empty-vs-valued comparison)"
);
assert_eq!(
valued
.compare(&empty, &desc)
.expect("compare must not fail"),
Ordering::Less,
"DESC: a valued clustering value must sort BEFORE the empty one"
);
}
#[test]
fn issue_10_desc_empty_vs_valued_in_merge_output_order() {
let schema = schema_one_clustering("ck", "text", ClusteringOrder::Desc);
let merger = empty_merger(schema);
let pk = DecoratedKey::new(7, vec![0, 0, 0, 7]);
const TS: i64 = 1_700_000_000_000_000;
let make = |ck: &str| {
MergeEntry::new(
0,
pk.clone(),
Some(ck_text("ck", ck)),
TS,
RowData::Live {
cells: vec![CellData {
column: "v".to_string(),
value: Value::text(format!("row-{ck}")),
timestamp: TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}],
},
)
};
let input = vec![make(""), make("b"), make("a")];
let merged = merger
.merge_partition_rows(input)
.expect("merge_partition_rows must not fail");
let order: Vec<String> = merged
.iter()
.map(|e| match &e.clustering_key {
Some(ck) => match &ck.columns[0].1 {
Value::Text(s) => String::from_utf8_lossy(s).into_owned(),
other => format!("{other:?}"),
},
None => "<none>".to_string(),
})
.collect();
assert_eq!(
order,
vec!["b".to_string(), "a".to_string(), "".to_string()],
"DESC merge output: valued rows descending, empty clustering value LAST"
);
}
#[test]
fn issue_13_tombstone_beats_expiring_at_equal_ts() {
let schema = schema_one_clustering("ck", "text", ClusteringOrder::Asc);
let merger = empty_merger(schema);
let pk = DecoratedKey::new(11, vec![0, 0, 0, 11]);
const TS: i64 = 1_700_000_000_000_000;
let ck_cell = || CellData {
column: "ck".to_string(),
value: Value::text("c".to_string()),
timestamp: TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let expiring = MergeEntry::new(
0,
pk.clone(),
Some(ck_text("ck", "c")),
TS,
RowData::Live {
cells: vec![
ck_cell(),
CellData {
column: "v".to_string(),
value: Value::text("expiring-if-buggy".to_string()),
timestamp: TS,
ttl: Some(3600),
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let tombstone = MergeEntry::new(
1,
pk.clone(),
Some(ck_text("ck", "c")),
TS,
RowData::Live {
cells: vec![
ck_cell(),
CellData {
column: "v".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: TS,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: 0,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let merged = merger
.merge_partition_rows(vec![expiring, tombstone])
.expect("merge_partition_rows must not fail");
assert_eq!(merged.len(), 1, "one clustering key => one merged winner");
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live row, got {other:?}"),
};
let v_cell = cells
.iter()
.find(|c| c.column == "v")
.expect("column v must survive (as a tombstone)");
assert!(
KWayMerger::is_cell_tombstone(v_cell),
"At equal ts the cell tombstone must beat the expiring (TTL) cell, even \
though the expiring cell is in the newer file (run 0). Got a live value \
=> tombstone-vs-expiring tie reverted to recency (#13/#3 regression)."
);
assert!(
v_cell.ttl.is_none(),
"Surviving cell tombstone must carry no TTL (IS_DELETED and IS_EXPIRING \
are mutually exclusive)."
);
}
#[test]
fn issue_13_tombstone_beats_expiring_irrespective_of_run_index() {
let schema = schema_one_clustering("ck", "text", ClusteringOrder::Asc);
let merger = empty_merger(schema);
let pk = DecoratedKey::new(12, vec![0, 0, 0, 12]);
const TS: i64 = 1_700_000_000_000_000;
let ck_cell = || CellData {
column: "ck".to_string(),
value: Value::text("c".to_string()),
timestamp: TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
let tombstone = MergeEntry::new(
0,
pk.clone(),
Some(ck_text("ck", "c")),
TS,
RowData::Live {
cells: vec![
ck_cell(),
CellData {
column: "v".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: TS,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: 0,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: TS,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let expiring = MergeEntry::new(
1,
pk.clone(),
Some(ck_text("ck", "c")),
TS,
RowData::Live {
cells: vec![
ck_cell(),
CellData {
column: "v".to_string(),
value: Value::text("expiring-if-buggy".to_string()),
timestamp: TS,
ttl: Some(3600),
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
},
],
},
);
let merged = merger
.merge_partition_rows(vec![tombstone, expiring])
.expect("merge_partition_rows must not fail");
let cells = match &merged[0].row_data {
RowData::Live { cells } => cells,
other => panic!("expected Live row, got {other:?}"),
};
let v_cell = cells.iter().find(|c| c.column == "v").expect("v survives");
assert!(
KWayMerger::is_cell_tombstone(v_cell),
"Tombstone must win the equal-ts tie over an expiring cell regardless of \
run_index ordering."
);
}
#[test]
fn issue_3_tombstone_beats_expiring_and_writer_never_sets_both_flags() {
let schema = TableSchema {
keyspace: "issue_822".to_string(),
table: "tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: ClusteringOrder::Asc,
}],
columns: vec![Column {
name: "v".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
const TS: i64 = 2_000_000;
let key = DecoratedKey::new(1, int_key_bytes(1));
let expiring = Mutation::new(
TableId::new("issue_822", "tbl"),
PartitionKey::single("pk", Value::Integer(1)),
Some(ClusteringKey::single("ck", Value::Integer(7))),
vec![CellOperation::WriteWithTtl {
column: "v".to_string(),
value: Value::text("expiring-if-buggy".to_string()),
ttl_seconds: 3600,
local_deletion_time: None,
}],
TS,
None,
);
let tombstone = Mutation::new(
TableId::new("issue_822", "tbl"),
PartitionKey::single("pk", Value::Integer(1)),
Some(ClusteringKey::single("ck", Value::Integer(7))),
vec![CellOperation::Delete {
column: "v".to_string(),
local_deletion_time: None,
}],
TS,
None,
);
let mut w = DataWriter::new(writer_stats());
w.write_partition(&key, &[expiring, tombstone], &schema, None, &[])
.expect("write_partition must succeed");
let bytes = w.finish().expect("finish must succeed");
let mut p = INT_PK_HEADER_SIZE;
let row_flags = bytes[p];
p += 1;
assert_eq!(
row_flags & ROW_HAS_EXTENDED_FLAGS,
0,
"regular (non-static) row expected"
);
assert_ne!(
row_flags & ROW_HAS_TIMESTAMP,
0,
"row keeps liveness ts from the expiring write"
);
assert_eq!(
row_flags & ROW_HAS_ALL_COLUMNS,
0,
"a surviving cell tombstone forces a column subset/bitmap (NOT all-columns)"
);
p += 1 + 4;
let (_row_size, rs_len) = read_vuint(&bytes, p);
p += rs_len;
let (_prev_size, ps_len) = read_vuint(&bytes, p);
p += ps_len;
let (_ts_delta, ts_len) = read_vuint(&bytes, p);
p += ts_len;
let (_bitmap, bm_len) = read_vuint(&bytes, p);
p += bm_len;
let cell_flags = bytes[p];
assert_ne!(
cell_flags & CELL_IS_DELETED,
0,
"At equal ts the cell tombstone must win and be serialized as a deleted \
cell (CELL_IS_DELETED set). Flags byte = {cell_flags:#04x}"
);
assert_eq!(
cell_flags & CELL_IS_EXPIRING,
0,
"A tombstone cell must NOT carry CELL_IS_EXPIRING — IS_DELETED and \
IS_EXPIRING are mutually exclusive. Flags byte = {cell_flags:#04x}"
);
assert_ne!(
cell_flags & (CELL_IS_DELETED | CELL_IS_EXPIRING),
CELL_IS_DELETED | CELL_IS_EXPIRING,
"the writer must never set BOTH CELL_IS_DELETED and CELL_IS_EXPIRING on \
one cell. Flags byte = {cell_flags:#04x}"
);
assert_eq!(
cell_flags, 0x05,
"surviving cell tombstone must serialize as CELL_IS_DELETED|HAS_EMPTY \
(0x05); got {cell_flags:#04x}"
);
}
#[test]
fn issue_848_writer_tombstone_beats_expiring_tombstone_first() {
let schema = TableSchema {
keyspace: "issue_822".to_string(),
table: "tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: ClusteringOrder::Asc,
}],
columns: vec![Column {
name: "v".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
const TS: i64 = 2_000_000;
let key = DecoratedKey::new(1, int_key_bytes(1));
let tombstone = Mutation::new(
TableId::new("issue_822", "tbl"),
PartitionKey::single("pk", Value::Integer(1)),
Some(ClusteringKey::single("ck", Value::Integer(7))),
vec![CellOperation::Delete {
column: "v".to_string(),
local_deletion_time: None,
}],
TS,
None,
);
let expiring = Mutation::new(
TableId::new("issue_822", "tbl"),
PartitionKey::single("pk", Value::Integer(1)),
Some(ClusteringKey::single("ck", Value::Integer(7))),
vec![CellOperation::WriteWithTtl {
column: "v".to_string(),
value: Value::text("expiring-if-buggy".to_string()),
ttl_seconds: 3600,
local_deletion_time: None,
}],
TS,
None,
);
let mut w = DataWriter::new(writer_stats());
w.write_partition(&key, &[tombstone, expiring], &schema, None, &[])
.expect("write_partition must succeed");
let bytes = w.finish().expect("finish must succeed");
let mut p = INT_PK_HEADER_SIZE;
let row_flags = bytes[p];
p += 1;
assert_eq!(
row_flags & ROW_HAS_ALL_COLUMNS,
0,
"a surviving cell tombstone forces a column subset/bitmap"
);
p += 1 + 4;
let (_row_size, rs_len) = read_vuint(&bytes, p);
p += rs_len;
let (_prev_size, ps_len) = read_vuint(&bytes, p);
p += ps_len;
let (_ts_delta, ts_len) = read_vuint(&bytes, p);
p += ts_len;
let (_bitmap, bm_len) = read_vuint(&bytes, p);
p += bm_len;
let cell_flags = bytes[p];
assert_ne!(
cell_flags & CELL_IS_DELETED,
0,
"tombstone-first: at equal ts the cell tombstone must still win \
(CELL_IS_DELETED set). Flags byte = {cell_flags:#04x}"
);
assert_eq!(
cell_flags & CELL_IS_EXPIRING,
0,
"the surviving tombstone must NOT carry CELL_IS_EXPIRING. \
Flags byte = {cell_flags:#04x}"
);
assert_eq!(
cell_flags, 0x05,
"surviving cell tombstone serializes as CELL_IS_DELETED|HAS_EMPTY \
(0x05); got {cell_flags:#04x}"
);
}
#[test]
fn issue_3_merge_layer_tombstone_carries_no_ttl_precondition() {
let tomb = CellData {
column: "v".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: 1,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: 0,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: 1,
ttl: None,
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
assert!(KWayMerger::is_cell_tombstone(&tomb));
assert!(
tomb.ttl.is_none(),
"A cell tombstone must not carry a TTL (precondition for flag exclusivity)."
);
let expiring = CellData {
column: "v".to_string(),
value: Value::text("x".to_string()),
timestamp: 1,
ttl: Some(60),
cell_path: None,
local_deletion_time: None,
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
};
assert!(
!KWayMerger::is_cell_tombstone(&expiring),
"An expiring (TTL) cell is LIVE, not a tombstone."
);
assert!(expiring.ttl.is_some(), "Expiring cell carries a TTL.");
}
#[test]
fn issue_22_static_emission_is_schema_driven_not_header_driven_divergent() {
let schema_with_static = TableSchema {
keyspace: "issue_822".to_string(),
table: "tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: ClusteringOrder::Asc,
}],
columns: vec![
Column {
name: "s".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: true,
},
Column {
name: "v".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
};
let mut schema_dropped_static = schema_with_static.clone();
schema_dropped_static.columns.retain(|c| c.name != "s");
let key = DecoratedKey::new(1, int_key_bytes(1));
let mutation = || {
Mutation::new(
TableId::new("issue_822", "tbl"),
PartitionKey::single("pk", Value::Integer(1)),
Some(ClusteringKey::single("ck", Value::Integer(7))),
vec![
CellOperation::Write {
column: "s".to_string(),
value: Value::text("static-val".to_string()),
},
CellOperation::Write {
column: "v".to_string(),
value: Value::text("row-val".to_string()),
},
],
2_000_000,
None,
)
};
const INT_CLUSTERING_PREFIX_LEN: usize = 1 + 4;
let walk_unfiltered = |bytes: &[u8], pos: usize| -> (u8, Option<u8>, usize) {
let mut p = pos;
let flags = bytes[p];
p += 1;
let mut ext = None;
if flags & ROW_HAS_EXTENDED_FLAGS != 0 {
ext = Some(bytes[p]);
p += 1;
}
let is_static = ext.is_some_and(|e| e & EXTENDED_IS_STATIC != 0);
if !is_static {
p += INT_CLUSTERING_PREFIX_LEN;
}
let (row_size, rs_len) = read_vuint(bytes, p);
p += rs_len;
let next = p + row_size as usize;
(flags, ext, next)
};
let first_unfiltered_is_static = |bytes: &[u8]| -> bool {
let (flags, ext, _next) = walk_unfiltered(bytes, INT_PK_HEADER_SIZE);
flags & ROW_HAS_EXTENDED_FLAGS != 0 && ext.is_some_and(|e| e & EXTENDED_IS_STATIC != 0)
};
let mut w_with = DataWriter::new(writer_stats());
w_with
.write_partition(&key, &[mutation()], &schema_with_static, None, &[])
.expect("write_partition (with static) must succeed");
let bytes_with = w_with.finish().expect("finish (with static)");
assert!(
first_unfiltered_is_static(&bytes_with),
"schema WITH a static column: the writer must emit a static-row prelude \
(ROW_HAS_EXTENDED_FLAGS | EXTENDED_IS_STATIC) as the first unfiltered"
);
let mut w_drop = DataWriter::new(writer_stats());
w_drop
.write_partition(&key, &[mutation()], &schema_dropped_static, None, &[])
.expect("write_partition (dropped static) must succeed");
let bytes_drop = w_drop.finish().expect("finish (dropped static)");
assert!(
!first_unfiltered_is_static(&bytes_drop),
"DIVERGENT (#22): with the static column dropped from the SCHEMA, CQLite's \
writer emits NO static-row prelude — even though a real cluster's on-disk \
SerializationHeader still records the static column and Cassandra \
(header-driven) would still emit its static rows. The writer's decision \
is schema-driven, observed here on the real Data.db bytes. If CQLite \
becomes header-driven this assertion must change."
);
const END_OF_PARTITION: u8 = 0x01;
let mut p = INT_PK_HEADER_SIZE;
while p < bytes_drop.len() {
if bytes_drop[p] == END_OF_PARTITION {
break;
}
let (flags, ext, next) = walk_unfiltered(&bytes_drop, p);
if flags & ROW_HAS_EXTENDED_FLAGS != 0 {
assert_eq!(
ext.expect("HAS_EXTENDED_FLAGS implies an ext-flags byte") & EXTENDED_IS_STATIC,
0,
"no EXTENDED_IS_STATIC row may appear once the static column is \
dropped from the schema"
);
}
assert!(next > p, "unfiltered walk must advance");
p = next;
}
let _ = ROW_HAS_DELETION;
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_886_empty_partition_skip {
use super::*;
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::{DecoratedKey, PartitionKey};
use std::collections::HashMap;
fn schema() -> TableSchema {
TableSchema {
keyspace: "i886".to_string(),
table: "phantom".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![crate::schema::Column {
name: "name".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn pk_bytes(schema: &TableSchema, n: i32) -> Vec<u8> {
PartitionKey::single("id", Value::Integer(n))
.to_bytes(schema)
.expect("encode int partition key")
}
struct VecIterator(std::vec::IntoIter<MergeEntry>);
impl SSTableRowIterator for VecIterator {
fn next(&mut self) -> Option<Result<MergeEntry>> {
self.0.next().map(Ok)
}
}
fn merger_over(entries: Vec<MergeEntry>, schema: TableSchema) -> KWayMerger {
KWayMerger {
runs: vec![RunReader::new(Box::new(VecIterator(entries.into_iter())))],
heap: BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema_arc: std::sync::Arc::new(schema.clone()),
schema,
}
}
#[tokio::test]
async fn empty_partition_is_skipped_on_writer_path() {
let schema = schema();
let meta_only = MergeEntry::new(
0,
DecoratedKey::new(1, pk_bytes(&schema, 1)),
None,
0,
RowData::Live { cells: vec![] },
);
assert!(
meta_only.is_metadata_only_no_op(),
"test precondition: the token-1 entry must be a metadata-only no-op"
);
let live = MergeEntry::new(
0,
DecoratedKey::new(2, pk_bytes(&schema, 2)),
None,
100,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("survivor".to_string()),
100,
)],
},
);
let merger = merger_over(vec![meta_only, live], schema.clone());
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
temp_dir.path().to_path_buf(),
1,
&schema,
)
.expect("create writer");
writer.pre_seed_encoding_baselines(0, i32::MAX, i32::MAX);
let stats = merger.merge(&mut writer).expect("merge must succeed");
assert_eq!(
stats.output_partitions, 1,
"only the normal partition counts as an output partition (no phantom)"
);
assert_eq!(
stats.output_rows, 1,
"only the normal partition's single live row counts toward output rows"
);
let info = writer.finish().await.expect("finish must succeed");
assert_eq!(
info.partition_count, 1,
"the output SSTable must contain exactly ONE partition; a phantom EMPTY \
partition for the metadata-only-only key would make this 2"
);
}
#[tokio::test]
async fn partition_with_real_content_is_still_written() {
let schema = schema();
let live = MergeEntry::new(
0,
DecoratedKey::new(7, pk_bytes(&schema, 7)),
None,
200,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("keep-me".to_string()),
200,
)],
},
);
let merger = merger_over(vec![live], schema.clone());
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
temp_dir.path().to_path_buf(),
1,
&schema,
)
.expect("create writer");
writer.pre_seed_encoding_baselines(0, i32::MAX, i32::MAX);
let stats = merger.merge(&mut writer).expect("merge must succeed");
assert_eq!(stats.output_partitions, 1);
assert_eq!(stats.output_rows, 1);
let info = writer.finish().await.expect("finish must succeed");
assert_eq!(
info.partition_count, 1,
"a normal partition must still be written"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_912_row_tombstone_clustering_identity {
use super::*;
use crate::schema::{ClusteringColumn, Column, KeyColumn, TableSchema};
use crate::storage::sstable::reader::compaction_row::{
CompactionRow, CompactionRowData, SimpleCell,
};
use crate::types::RowKey;
use std::collections::HashMap;
fn clustered_schema() -> TableSchema {
TableSchema {
keyspace: "ks912".to_string(),
table: "t912".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order: Default::default(),
}],
columns: vec![Column {
name: "v".to_string(),
data_type: "int".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn row_tombstone(ck: i64) -> CompactionRowData {
CompactionRowData::Tombstone {
deletion_time: 1_000 + ck,
local_deletion_time: 42,
clustering: vec![("ck".to_string(), Value::Integer(ck as i32))],
}
}
#[test]
fn tombstone_clustering_identity_is_distinct() {
let schema = clustered_schema();
let ck5 = SSTableRowIteratorAdapter::extract_clustering_key_from_compaction(
&row_tombstone(5),
&schema,
);
let ck9 = SSTableRowIteratorAdapter::extract_clustering_key_from_compaction(
&row_tombstone(9),
&schema,
);
assert_eq!(
ck5,
Some(ClusteringKey {
columns: vec![("ck".to_string(), Value::Integer(5))],
}),
"row tombstone must carry its clustering identity (#912)"
);
assert_ne!(
ck5, ck9,
"distinct clustering-row tombstones must not share a bucket"
);
}
#[test]
fn tombstone_without_clustering_falls_into_none_bucket() {
let schema = clustered_schema();
let bare = CompactionRowData::Tombstone {
deletion_time: 1,
local_deletion_time: 0,
clustering: Vec::new(),
};
assert_eq!(
SSTableRowIteratorAdapter::extract_clustering_key_from_compaction(&bare, &schema),
None
);
}
#[test]
fn live_and_tombstone_same_ck_share_bucket() {
let schema = clustered_schema();
let live = CompactionRowData::Live {
simple: vec![SimpleCell {
column: "ck".to_string(),
value: Value::Integer(5),
timestamp: 10,
ttl: None,
local_deletion_time: None,
}],
complex: Vec::new(),
row_deletion: None,
};
let live_ck =
SSTableRowIteratorAdapter::extract_clustering_key_from_compaction(&live, &schema);
let tomb_ck = SSTableRowIteratorAdapter::extract_clustering_key_from_compaction(
&row_tombstone(5),
&schema,
);
assert_eq!(
live_ck, tomb_ck,
"same ck => same bucket for live row and its tombstone"
);
assert!(live_ck.is_some());
}
#[test]
fn two_row_tombstones_do_not_collapse_in_merge() {
let schema = clustered_schema();
let pk = RowKey::new(vec![0, 0, 0, 7]);
let e5 = SSTableRowIteratorAdapter::build_merge_entry(
0,
CompactionRow {
key: pk.clone(),
row_timestamp: 1_005,
row_data: row_tombstone(5),
},
&schema,
)
.expect("build_merge_entry");
let e9 = SSTableRowIteratorAdapter::build_merge_entry(
1,
CompactionRow {
key: pk.clone(),
row_timestamp: 1_009,
row_data: row_tombstone(9),
},
&schema,
)
.expect("build_merge_entry");
assert_ne!(
e5.clustering_key, e9.clustering_key,
"two distinct row tombstones must land in distinct buckets (#912)"
);
let merger = KWayMerger {
runs: vec![],
heap: std::collections::BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema: schema.clone(),
schema_arc: std::sync::Arc::new(schema.clone()),
};
let merged = merger
.merge_partition_rows(vec![e5, e9])
.expect("merge_partition_rows");
assert_eq!(
merged.len(),
2,
"both clustering-row tombstones must survive; pre-#912 they collapsed to one"
);
assert!(
merged
.iter()
.all(|m| matches!(m.row_data, RowData::Tombstone { .. })),
"both surviving entries must be row tombstones"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_873_preserve_row_tombstone_ldt {
use super::*;
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::DecoratedKey;
use std::collections::HashMap;
fn unclustered_schema() -> TableSchema {
TableSchema {
keyspace: "ks873".to_string(),
table: "t873".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn tombstone_entry(run_index: usize, deletion_time: i64, ldt: i32) -> MergeEntry {
MergeEntry::new(
run_index,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
deletion_time,
RowData::Tombstone {
deletion_time,
local_deletion_time: ldt,
},
)
}
#[test]
fn reconcile_cluster_preserves_winning_tombstone_ldt() {
let deletion_time = 1_000_000_000_000_000i64; let source_ldt = 1_700_000_000i32;
let merged = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, deletion_time, source_ldt)],
&HashMap::new(),
None,
)
.expect("a surviving row tombstone must be emitted");
match merged.row_data {
RowData::Tombstone {
deletion_time: dt,
local_deletion_time,
} => {
assert_eq!(dt, deletion_time, "deletion_time must survive");
assert_eq!(
local_deletion_time, source_ldt,
"the source LDT must be preserved, not reset to 0 nor derived from the timestamp"
);
}
other => panic!("expected Tombstone, got {:?}", other),
}
}
#[test]
fn reconcile_cluster_keeps_ldt_paired_with_max_deletion_time() {
let older = tombstone_entry(0, 100_000_000, 1_900_000_000);
let newer = tombstone_entry(1, 200_000_000, 1_500_000_000);
let merged = KWayMerger::reconcile_cluster(None, vec![older, newer], &HashMap::new(), None)
.expect("a surviving row tombstone must be emitted");
match merged.row_data {
RowData::Tombstone {
deletion_time,
local_deletion_time,
} => {
assert_eq!(deletion_time, 200_000_000, "the newer delete must win");
assert_eq!(
local_deletion_time, 1_500_000_000,
"the LDT carried must be the one paired with the winning deletion_time"
);
}
other => panic!("expected Tombstone, got {:?}", other),
}
}
#[test]
fn merge_entry_to_mutation_threads_row_tombstone_ldt() {
let schema = unclustered_schema();
let deletion_time = 1_000_000_000_000_000i64;
let source_ldt = 1_700_000_000i32;
let mutation = KWayMerger::merge_entry_to_mutation(
tombstone_entry(0, deletion_time, source_ldt),
&schema,
)
.expect("conversion should succeed");
assert_eq!(
mutation.local_deletion_time,
Some(source_ldt),
"the row tombstone's source LDT must be threaded onto the mutation"
);
assert_eq!(
mutation.effective_local_deletion_time(),
source_ldt,
"the writer must use the preserved LDT, not the timestamp-derived one"
);
assert_ne!(
mutation.effective_local_deletion_time() as i64,
deletion_time / 1_000_000,
"the LDT must NOT be re-derived from the deletion timestamp"
);
}
#[test]
fn merge_entry_to_mutation_placeholder_ldt_stays_unset() {
let schema = unclustered_schema();
let deletion_time = 1_000_000_000_000_000i64;
let mutation = KWayMerger::merge_entry_to_mutation(
tombstone_entry(0, deletion_time, 0),
&schema,
)
.expect("conversion should succeed");
assert_eq!(
mutation.local_deletion_time, None,
"a placeholder (0) LDT must leave the mutation LDT unset so the writer \
keeps deriving it from the timestamp, exactly as before #873"
);
}
#[test]
fn merge_entry_to_mutation_live_row_leaves_ldt_none() {
let schema = unclustered_schema();
let entry = MergeEntry::new(
0,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
100,
RowData::Live {
cells: vec![CellData::new(
"value".to_string(),
Value::text("alive".to_string()),
100,
)],
},
);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert_eq!(
mutation.local_deletion_time, None,
"a live row must not pin an explicit LDT"
);
}
#[test]
fn reconcile_cluster_attaches_row_deletion_when_cells_survive() {
let deletion_time = 100i64;
let source_ldt = 1_700_000_000i32;
let tomb = tombstone_entry(1, deletion_time, source_ldt);
let live = MergeEntry::new(
0,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
300,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("new".to_string()),
300,
)],
},
);
let merged = KWayMerger::reconcile_cluster(None, vec![tomb, live], &HashMap::new(), None)
.expect("a live coexistence row must be emitted");
match &merged.row_data {
RowData::Live { cells } => {
assert_eq!(cells.len(), 1, "the surviving name cell must remain");
assert_eq!(cells[0].column, "name");
assert_eq!(cells[0].timestamp, 300);
}
other => panic!("expected a Live coexistence row, got {:?}", other),
}
assert_eq!(
merged.row_deletion,
Some((deletion_time, source_ldt)),
"the coexisting row deletion (and its source LDT) must be preserved on the live entry"
);
}
#[test]
fn reconcile_cluster_shadows_older_cell_under_coexisting_deletion() {
let deletion_time = 100i64;
let tomb = tombstone_entry(1, deletion_time, 1_700_000_000);
let older = MergeEntry::new(
2,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
50,
RowData::Live {
cells: vec![CellData::new("score".to_string(), Value::Integer(999), 50)],
},
);
let newer = MergeEntry::new(
0,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
300,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("new".to_string()),
300,
)],
},
);
let merged =
KWayMerger::reconcile_cluster(None, vec![tomb, older, newer], &HashMap::new(), None)
.expect("a live coexistence row must be emitted");
match &merged.row_data {
RowData::Live { cells } => {
assert_eq!(cells.len(), 1, "only the newer cell survives");
assert_eq!(
cells[0].column, "name",
"the older `score` cell is shadowed"
);
}
other => panic!("expected a Live coexistence row, got {:?}", other),
}
assert_eq!(
merged.row_deletion.map(|(dt, _)| dt),
Some(deletion_time),
"the row deletion must be preserved to keep shadowing across SSTables"
);
}
#[test]
fn merge_entry_to_mutation_emits_coexisting_row_tombstone() {
let schema = unclustered_schema();
let deletion_time = 100i64;
let source_ldt = 1_700_000_000i32;
let entry = MergeEntry::new(
0,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
300,
RowData::Live {
cells: vec![CellData::new(
"name".to_string(),
Value::text("new".to_string()),
300,
)],
},
)
.with_row_deletion(deletion_time, source_ldt);
let mutation =
KWayMerger::merge_entry_to_mutation(entry, &schema).expect("conversion should succeed");
assert_eq!(
mutation.row_tombstone,
Some((deletion_time, source_ldt)),
"the coexisting row deletion must be emitted as Mutation::row_tombstone"
);
assert_eq!(
mutation.timestamp_micros, 300,
"the mutation's liveness timestamp stays the row write time, NOT the deletion time"
);
assert!(
mutation
.operations
.iter()
.any(|op| matches!(op, crate::storage::write_engine::mutation::CellOperation::Write { column, .. } if column == "name")),
"the surviving cell must still be emitted alongside the row tombstone"
);
}
struct VecIterator(std::vec::IntoIter<MergeEntry>);
impl SSTableRowIterator for VecIterator {
fn next(&mut self) -> Option<Result<MergeEntry>> {
self.0.next().map(Ok)
}
}
#[tokio::test]
async fn row_tombstone_ldt_survives_compaction_rewrite() {
use crate::platform::Platform;
use crate::storage::sstable::reader::compaction_row::CompactionRowData;
use crate::Config;
use std::sync::Arc;
let schema = unclustered_schema();
let deletion_time = 1_000_000_000_000_000i64; let source_ldt = 1_700_000_000i32;
let merger = KWayMerger {
runs: vec![RunReader::new(Box::new(VecIterator(
vec![tombstone_entry(0, deletion_time, source_ldt)].into_iter(),
)))],
heap: std::collections::BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema: schema.clone(),
schema_arc: std::sync::Arc::new(schema.clone()),
};
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
temp_dir.path().to_path_buf(),
1,
&schema,
)
.expect("create writer");
writer.pre_seed_encoding_baselines(deletion_time, source_ldt, i32::MAX);
merger
.merge(&mut writer)
.expect("merge+write must succeed (no LDT underflow)");
let info = writer.finish().await.expect("finish must succeed");
let config = Config::default();
let platform = Arc::new(Platform::new(&config).await.expect("platform"));
let reader = crate::storage::sstable::reader::SSTableReader::open(
&info.data_path,
&config,
platform,
)
.await
.expect("open written SSTable");
let rows = reader
.iterate_all_partitions_for_compaction(Some(&schema))
.await
.expect("compaction iterator");
let mut saw_tombstone = false;
for row in rows {
if let CompactionRowData::Tombstone {
deletion_time: dt,
local_deletion_time,
..
} = row.row_data
{
saw_tombstone = true;
assert_eq!(dt, deletion_time, "deletion_time must round-trip");
assert_eq!(
local_deletion_time, source_ldt,
"local_deletion_time must round-trip the preserved source value"
);
}
}
assert!(
saw_tombstone,
"the rewritten SSTable must contain the row tombstone"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_845_gc_grace_purge {
use super::*;
use crate::schema::{KeyColumn, TableSchema};
use crate::storage::write_engine::mutation::{ClusteringBound, DecoratedKey};
use crate::types::{TombstoneInfo, TombstoneType};
use std::collections::HashMap;
fn dk(byte: u8) -> DecoratedKey {
DecoratedKey::from_key_bytes(vec![byte]).expect("token")
}
fn live(run_index: usize, row_ts: i64, cells: Vec<CellData>) -> MergeEntry {
MergeEntry::new(run_index, dk(1), None, row_ts, RowData::Live { cells })
}
fn unclustered_schema() -> TableSchema {
TableSchema {
keyspace: "ks845".to_string(),
table: "t845".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn tombstone_entry(run_index: usize, deletion_time: i64, ldt: i32) -> MergeEntry {
MergeEntry::new(
run_index,
DecoratedKey::new(500, 7i32.to_be_bytes().to_vec()),
None,
deletion_time,
RowData::Tombstone {
deletion_time,
local_deletion_time: ldt,
},
)
}
fn cell_tombstone(ts: i64, ldt: i32) -> CellData {
CellData {
column: "v".to_string(),
value: Value::Tombstone(Box::new(TombstoneInfo {
deletion_time: ts,
tombstone_type: TombstoneType::CellTombstone,
local_deletion_time: ldt as i64,
ttl: None,
range_start: None,
range_end: None,
})),
timestamp: ts,
ttl: None,
cell_path: None,
local_deletion_time: Some(ldt),
is_complex_element: false,
is_deleted: false,
has_empty_value: false,
}
}
#[test]
fn issue_845_complex_deletion_purged_when_older_than_gc_before() {
const GC_BEFORE: i64 = 1_700_000_000;
let make = |ldt: i32| {
MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 50,
local_deletion_time: ldt,
}])
};
let purged = KWayMerger::reconcile_cluster(
None,
vec![make((GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
purged.is_none(),
"a complex-deletion marker older than gcBefore must be purged, \
leaving nothing to emit"
);
let retained = KWayMerger::reconcile_cluster(
None,
vec![make((GC_BEFORE + 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a within-grace marker must keep a metadata-only entry");
assert_eq!(
retained.complex_deletions.len(),
1,
"a complex-deletion marker within grace must be retained"
);
let boundary = KWayMerger::reconcile_cluster(
None,
vec![make(GC_BEFORE as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a marker at exactly gcBefore must be retained");
assert_eq!(
boundary.complex_deletions.len(),
1,
"localDeletionTime == gcBefore is within grace (only `<` purges)"
);
}
#[test]
fn issue_845_row_tombstone_purged_when_older_than_gc_before() {
const GC_BEFORE: i64 = 1_700_000_000;
let dt = 1_000_000_000_000_000i64;
let older = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
older.is_none(),
"a row tombstone older than gcBefore must be purged (nothing emitted)"
);
let within = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, (GC_BEFORE + 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a within-grace row tombstone must be retained");
assert!(
matches!(within.row_data, RowData::Tombstone { .. }),
"a within-grace row tombstone must survive"
);
let boundary = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, GC_BEFORE as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a row tombstone at exactly gcBefore must be retained");
assert!(
matches!(boundary.row_data, RowData::Tombstone { .. }),
"localDeletionTime == gcBefore is within grace (only `<` purges)"
);
}
#[test]
fn issue_845_cell_tombstone_purged_when_older_than_gc_before() {
const GC_BEFORE: i64 = 1_700_000_000;
let keep = CellData::new("name".to_string(), Value::text("alive".to_string()), 500);
let count_tombstone_cells = |ldt: i32| -> usize {
let merged = KWayMerger::reconcile_cluster(
None,
vec![live(0, 500, vec![keep.clone(), cell_tombstone(100, ldt)])],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a live row must be emitted");
match merged.row_data {
RowData::Live { cells } => cells
.iter()
.filter(|c| KWayMerger::is_cell_tombstone(c))
.count(),
other => panic!("expected Live, got {other:?}"),
}
};
assert_eq!(
count_tombstone_cells((GC_BEFORE - 1) as i32),
0,
"a cell tombstone older than gcBefore must be purged from the row"
);
assert_eq!(
count_tombstone_cells((GC_BEFORE + 1) as i32),
1,
"a cell tombstone within grace must be retained"
);
assert_eq!(
count_tombstone_cells(GC_BEFORE as i32),
1,
"localDeletionTime == gcBefore is within grace (only `<` purges)"
);
}
#[test]
fn issue_845_no_gc_before_retains_everything() {
let dt = 1_000_000_000_000_000i64;
let merged = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, 1)],
&::std::collections::HashMap::new(),
None,
)
.expect("without gcBefore the tombstone must be retained");
assert!(
matches!(merged.row_data, RowData::Tombstone { .. }),
"with gc_before_secs = None nothing is purged"
);
}
#[test]
fn issue_845_purge_does_not_resurrect_shadowed_element() {
const GC_BEFORE: i64 = 1_700_000_000;
const MFDA: i64 = 200;
let covered = CellData {
column: "tags".to_string(),
value: Value::text("ghost".to_string()),
timestamp: MFDA,
ttl: None,
cell_path: Some(vec![1, 2, 3]),
local_deletion_time: None,
is_complex_element: true,
is_deleted: false,
has_empty_value: false,
};
let entry = MergeEntry::new(
0,
dk(1),
None,
MFDA,
RowData::Live {
cells: vec![covered],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: MFDA,
local_deletion_time: (GC_BEFORE - 1) as i32,
}]);
let merged = KWayMerger::reconcile_cluster(
None,
vec![entry],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
merged.is_none(),
"purging the marker must not resurrect the element it shadowed"
);
}
#[test]
fn issue_845_compute_gc_before_from_schema() {
let mut schema = unclustered_schema();
let now = 2_000_000_000i64;
schema
.comments
.insert("gc_grace_seconds".to_string(), "0".to_string());
assert_eq!(compute_gc_before(&schema, now), Some(now));
schema
.comments
.insert("gc_grace_seconds".to_string(), "864000".to_string());
assert_eq!(compute_gc_before(&schema, now), Some(now - 864_000));
schema
.comments
.insert("gc_grace_seconds".to_string(), "not-a-number".to_string());
assert_eq!(compute_gc_before(&schema, now), None);
}
#[test]
fn issue_921_compute_gc_before_defaults_to_864000_when_absent() {
let mut schema = unclustered_schema();
let now = 2_000_000_000i64;
assert_eq!(
compute_gc_before(&schema, now),
Some(now - 864_000),
"missing gc_grace_seconds must use the Cassandra default of 864000s"
);
schema
.comments
.insert("gc_grace_seconds".to_string(), "0".to_string());
assert_eq!(
compute_gc_before(&schema, now),
Some(now),
"gc_grace_seconds = 0 is valid (immediate grace)"
);
schema
.comments
.insert("gc_grace_seconds".to_string(), "-1".to_string());
assert_eq!(
compute_gc_before(&schema, now),
None,
"a negative gc_grace_seconds must disable purging (no-op)"
);
schema
.comments
.insert("gc_grace_seconds".to_string(), "garbage".to_string());
assert_eq!(
compute_gc_before(&schema, now),
None,
"an unparseable gc_grace_seconds must disable purging (no-op)"
);
}
#[test]
fn issue_921_unsigned_local_deletion_time_not_purged() {
let future_ldt_bits = 0x8000_0000u32 as i32;
assert!(future_ldt_bits < 0, "the wrapped LDT is a negative i32");
const GC_BEFORE: i64 = 1_700_000_000;
assert!(
i64::from(future_ldt_bits as u32) > GC_BEFORE,
"the unsigned LDT is in the future relative to gcBefore"
);
let dt = 1_000_000_000_000_000i64;
let row = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, future_ldt_bits)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a far-future row tombstone must NOT be purged");
assert!(
matches!(row.row_data, RowData::Tombstone { .. }),
"an unsigned far-future row tombstone must survive a normal gcBefore"
);
let complex = KWayMerger::reconcile_cluster(
None,
vec![
MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: 50,
local_deletion_time: future_ldt_bits,
}]),
],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a far-future complex-deletion marker must NOT be purged");
assert_eq!(
complex.complex_deletions.len(),
1,
"an unsigned far-future complex marker must survive a normal gcBefore"
);
let keep = CellData::new("name".to_string(), Value::text("alive".to_string()), 500);
let cell = KWayMerger::reconcile_cluster(
None,
vec![live(
0,
500,
vec![keep, cell_tombstone(100, future_ldt_bits)],
)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a live row must be emitted");
let tombstone_cells = match cell.row_data {
RowData::Live { cells } => cells
.iter()
.filter(|c| KWayMerger::is_cell_tombstone(c))
.count(),
other => panic!("expected Live, got {other:?}"),
};
assert_eq!(
tombstone_cells, 1,
"an unsigned far-future cell tombstone must survive a normal gcBefore"
);
let ancient = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
ancient.is_none(),
"a genuinely ancient row tombstone must still be purged"
);
}
#[test]
fn issue_921_partial_compaction_does_not_purge_but_major_does() {
const GC_BEFORE: i64 = 1_700_000_000;
let dt = 1_000_000_000_000_000i64;
let ancient_ldt = (GC_BEFORE - 1) as i32;
let merger = |purge_safe: bool| KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
schema: unclustered_schema(),
schema_arc: std::sync::Arc::new(unclustered_schema()),
gc_before_secs: Some(GC_BEFORE),
now_secs: None,
purge_safe,
max_purgeable_timestamp: None,
};
let partial = merger(false)
.merge_partition_rows(vec![tombstone_entry(0, dt, ancient_ldt)])
.expect("merge must succeed");
assert_eq!(
partial.len(),
1,
"a partial (non-overlap-safe) compaction must RETAIN the tombstone"
);
assert!(
matches!(partial[0].row_data, RowData::Tombstone { .. }),
"the retained entry must still be a row tombstone (no resurrection risk)"
);
let major = merger(true)
.merge_partition_rows(vec![tombstone_entry(0, dt, ancient_ldt)])
.expect("merge must succeed");
assert!(
major.is_empty(),
"a major (overlap-safe) compaction must PURGE the ancient tombstone"
);
}
#[test]
fn issue_1061_range_tombstone_purge_respects_overlap_bound() {
const GC_BEFORE: i64 = 1_700_000_000;
let ancient_ldt = (GC_BEFORE - 1) as i32;
const BOUND: i64 = 2_000;
let merger = |purge_safe: bool, bound: Option<i64>| KWayMerger {
runs: vec![],
heap: BinaryHeap::new(),
current_partition: None,
schema: unclustered_schema(),
schema_arc: std::sync::Arc::new(unclustered_schema()),
gc_before_secs: Some(GC_BEFORE),
now_secs: None,
purge_safe,
max_purgeable_timestamp: bound,
};
let carrier = |deletion_time: i64| {
let rt = RangeTombstone {
start: ClusteringBound::Bottom,
end: ClusteringBound::Top,
deletion_time,
local_deletion_time: ancient_ldt,
};
MergeEntry::new(
0,
dk(1),
None,
deletion_time,
RowData::Live { cells: vec![] },
)
.with_range_deletion(rt)
};
let at_bound = merger(false, Some(BOUND))
.merge_partition_rows(vec![carrier(BOUND)])
.expect("merge must succeed");
assert_eq!(
at_bound.len(),
1,
"a range tombstone AT the overlap bound must be RETAINED (#1061)"
);
assert!(
at_bound[0].range_deletion.is_some(),
"the retained entry must still carry the range-tombstone marker"
);
let above_bound = merger(false, Some(BOUND))
.merge_partition_rows(vec![carrier(BOUND + 1_000)])
.expect("merge must succeed");
assert_eq!(
above_bound.len(),
1,
"a range tombstone ABOVE the overlap bound must be RETAINED (#1061)"
);
let below_bound = merger(false, Some(BOUND))
.merge_partition_rows(vec![carrier(BOUND - 1)])
.expect("merge must succeed");
assert!(
below_bound.is_empty(),
"a gc-expired range tombstone strictly BELOW the overlap bound is purgeable"
);
let overlap_safe = merger(true, None)
.merge_partition_rows(vec![carrier(BOUND)])
.expect("merge must succeed");
assert!(
overlap_safe.is_empty(),
"an overlap-safe (major) compaction still purges the gc-expired marker"
);
}
#[test]
fn issue_921_row_tombstone_ldt_zero_is_unknown_and_retained() {
const GC_BEFORE: i64 = 1_700_000_000;
let dt = 1_000_000_000_000_000i64;
let unknown = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, 0)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a row tombstone with unknown (0) LDT must be RETAINED, not purged");
assert!(
matches!(unknown.row_data, RowData::Tombstone { .. }),
"LDT==0 is the unknown placeholder and must be retained (never purge \
on unknown LDT)"
);
let ancient = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
ancient.is_none(),
"a real non-zero ancient row tombstone (LDT < gcBefore) must still purge"
);
let within = KWayMerger::reconcile_cluster(
None,
vec![tombstone_entry(0, dt, (GC_BEFORE + 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a within-grace row tombstone must be retained");
assert!(
matches!(within.row_data, RowData::Tombstone { .. }),
"a within-grace row tombstone (LDT >= gcBefore) must survive"
);
}
#[test]
fn issue_921_clustered_row_with_only_purgeable_cell_tombstone_emits_nothing() {
const GC_BEFORE: i64 = 1_700_000_000;
let ck = ClusteringKey {
columns: vec![("ck".to_string(), Value::text("c1".to_string()))],
};
let ck_cell = CellData::new("ck".to_string(), Value::text("c1".to_string()), 100);
let make = |extra: Vec<CellData>| {
let mut cells = vec![ck_cell.clone()];
cells.extend(extra);
MergeEntry::new(0, dk(1), Some(ck.clone()), 100, RowData::Live { cells })
};
let purged = KWayMerger::reconcile_cluster(
Some(ck.clone()),
vec![make(vec![cell_tombstone(50, (GC_BEFORE - 1) as i32)])],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
);
assert!(
purged.is_none(),
"a clustered row whose only non-key data is a purgeable cell tombstone \
must emit NOTHING (no phantom key-only live row) after the gc purge"
);
let kept = KWayMerger::reconcile_cluster(
Some(ck.clone()),
vec![make(vec![
CellData::new("v".to_string(), Value::text("real".to_string()), 60),
cell_tombstone(50, (GC_BEFORE - 1) as i32),
])],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
)
.expect("a clustered row with real surviving data must emit a live row");
match kept.row_data {
RowData::Live { cells } => {
assert!(
cells
.iter()
.any(|c| c.column == "v" && !KWayMerger::is_cell_tombstone(c)),
"the real surviving `v` cell must remain in the emitted live row"
);
assert!(
!cells.iter().any(KWayMerger::is_cell_tombstone),
"the purgeable cell tombstone must still be purged from the live row"
);
}
other => panic!("expected Live row, got {other:?}"),
}
}
#[test]
fn issue_935_partial_compaction_purges_row_tombstone_below_overlap_bound() {
const GC_BEFORE: i64 = 1_700_000_000;
const BOUND: i64 = 1_000_000_000_000_000;
let mfda = BOUND - 1;
let purged = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![tombstone_entry(0, mfda, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
);
assert!(
purged.is_none(),
"a row tombstone older than every non-included overlapping SSTable \
(markedForDeleteAt < bound) and past grace must be purged even in a \
partial compaction (#935)"
);
}
#[test]
fn issue_935_partial_compaction_retains_row_tombstone_at_or_above_overlap_bound() {
const GC_BEFORE: i64 = 1_700_000_000;
const BOUND: i64 = 1_000_000_000_000_000;
let above = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![tombstone_entry(0, BOUND + 1, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("a tombstone that could shadow outside data must be retained");
assert!(
matches!(above.row_data, RowData::Tombstone { .. }),
"a row tombstone with markedForDeleteAt > overlap bound must survive a \
partial compaction (#935): outside data may be shadowed"
);
let boundary = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![tombstone_entry(0, BOUND, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("a tombstone at exactly the overlap bound must be retained");
assert!(
matches!(boundary.row_data, RowData::Tombstone { .. }),
"markedForDeleteAt == overlap bound is retained (only `<` purges, #935)"
);
}
#[test]
fn issue_935_within_grace_tombstone_retained_despite_overlap_bound() {
const GC_BEFORE: i64 = 1_700_000_000;
const BOUND: i64 = 1_000_000_000_000_000;
let within = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![tombstone_entry(0, BOUND - 1, (GC_BEFORE + 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("a within-grace tombstone must be retained regardless of overlap");
assert!(
matches!(within.row_data, RowData::Tombstone { .. }),
"a tombstone within gc grace must be retained even when its \
markedForDeleteAt is below the overlap bound (#935)"
);
}
#[test]
fn issue_935_partial_compaction_cell_tombstone_overlap_gate() {
const GC_BEFORE: i64 = 1_700_000_000;
const BOUND: i64 = 5_000;
let purged = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![live(
0,
100,
vec![
CellData::new("keep".to_string(), Value::text("x".to_string()), 9_000),
cell_tombstone(BOUND - 1, (GC_BEFORE - 1) as i32),
],
)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("the live `keep` cell keeps the row alive");
match purged.row_data {
RowData::Live { cells } => assert!(
!cells.iter().any(KWayMerger::is_cell_tombstone),
"a cell tombstone below the overlap bound must be purged (#935)"
),
other => panic!("expected Live row, got {other:?}"),
}
let retained = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![live(
0,
100,
vec![
CellData::new("keep".to_string(), Value::text("x".to_string()), 9_000),
cell_tombstone(BOUND, (GC_BEFORE - 1) as i32),
],
)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("the live `keep` cell keeps the row alive");
match retained.row_data {
RowData::Live { cells } => assert!(
cells.iter().any(KWayMerger::is_cell_tombstone),
"a cell tombstone at the overlap bound must be retained (#935)"
),
other => panic!("expected Live row, got {other:?}"),
}
}
#[test]
fn issue_935_partial_compaction_complex_deletion_overlap_gate() {
const GC_BEFORE: i64 = 1_700_000_000;
const BOUND: i64 = 5_000;
let make = |mfda: i64| {
MergeEntry::new(0, dk(1), None, 0, RowData::Live { cells: vec![] })
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: mfda,
local_deletion_time: (GC_BEFORE - 1) as i32,
}])
};
let purged = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![make(BOUND - 1)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
);
assert!(
purged.is_none(),
"a complex-deletion marker below the overlap bound must be purged (#935)"
);
let retained = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![make(BOUND)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
BOUND,
)
.expect("a marker at the overlap bound must be retained");
assert_eq!(
retained.complex_deletions.len(),
1,
"a complex-deletion marker at the overlap bound must be retained (#935)"
);
}
#[test]
fn issue_935_full_compaction_bound_is_unrestricted() {
const GC_BEFORE: i64 = 1_700_000_000;
let purged = KWayMerger::reconcile_cluster_with_overlap(
None,
vec![tombstone_entry(0, i64::MAX - 1, (GC_BEFORE - 1) as i32)],
&::std::collections::HashMap::new(),
Some(GC_BEFORE),
i64::MAX,
);
assert!(
purged.is_none(),
"the full-compaction +inf overlap bound must purge any past-grace \
tombstone, identical to #845 (#935 no-op for full compactions)"
);
}
#[tokio::test]
async fn issue_921_complex_deletion_marker_counts_as_dropped_column_survivor() {
use crate::schema::Column;
struct VecIterator(std::vec::IntoIter<MergeEntry>);
impl SSTableRowIterator for VecIterator {
fn next(&mut self) -> Option<Result<MergeEntry>> {
self.0.next().map(Ok)
}
}
fn schema_with_tags(dropped: HashMap<String, i64>) -> TableSchema {
TableSchema {
keyspace: "ks921".to_string(),
table: "t921".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "id".to_string(),
data_type: "int".to_string(),
nullable: false,
default: None,
is_static: false,
},
Column {
name: "tags".to_string(),
data_type: "set<text>".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: HashMap::new(),
dropped_columns: dropped,
}
}
const MARKER_LDT: i32 = 1_700_000_000;
const MARKER_MFDA: i64 = 1_600_000_000_000_000;
let write_schema = schema_with_tags(HashMap::new());
let entry = MergeEntry::new(
0,
DecoratedKey::new(7, 7i32.to_be_bytes().to_vec()),
None,
0,
RowData::Live { cells: vec![] },
)
.with_complex_deletions(vec![ComplexDeletion {
column: "tags".to_string(),
marked_for_delete_at: MARKER_MFDA,
local_deletion_time: MARKER_LDT,
}]);
let merger = KWayMerger {
runs: vec![RunReader::new(Box::new(VecIterator(
vec![entry].into_iter(),
)))],
heap: std::collections::BinaryHeap::new(),
current_partition: None,
gc_before_secs: None,
now_secs: None,
purge_safe: false,
max_purgeable_timestamp: None,
schema: write_schema.clone(),
schema_arc: std::sync::Arc::new(write_schema.clone()),
};
let temp_dir = tempfile::TempDir::new().expect("temp dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
temp_dir.path().to_path_buf(),
1,
&write_schema,
)
.expect("create writer");
writer.pre_seed_encoding_baselines(MARKER_MFDA, MARKER_LDT, i32::MAX);
merger.merge(&mut writer).expect("merge+write must succeed");
let info = writer.finish().await.expect("finish must succeed");
let data_path = info.data_path;
let mut dropped = HashMap::new();
dropped.insert("tags".to_string(), 1_i64);
let drop_schema = schema_with_tags(dropped);
let retained = compute_surviving_dropped_columns(
vec![data_path.clone()],
&drop_schema,
None,
None,
false,
None,
)
.expect("survivor pre-pass (no gc) must succeed");
assert!(
retained.contains("tags"),
"a RETAINED complex-deletion marker for a dropped complex column must \
count as a survivor so the writer keeps the column to emit it; got {retained:?}"
);
let gc_before = i64::from(MARKER_LDT) + 1;
let purged = compute_surviving_dropped_columns(
vec![data_path],
&drop_schema,
Some(gc_before),
Some(gc_before),
true,
None,
)
.expect("survivor pre-pass (purging) must succeed");
assert!(
!purged.contains("tags"),
"a PURGED complex-deletion marker must NOT count as a survivor (the \
dropped column is stripped from the output schema); got {purged:?}"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_929_bare_udt_compaction {
use super::*;
use crate::schema::{Column, CqlType, KeyColumn, UdtRegistry};
use crate::storage::write_engine::mutation::{CellOperation, Mutation, PartitionKey, TableId};
use crate::types::{UdtField, UdtTypeDef, UdtValue, Value};
fn person_registry() -> UdtRegistry {
let mut reg = UdtRegistry::new();
reg.register_udt(
UdtTypeDef::new("test_ks".to_string(), "person".to_string())
.with_field("name".to_string(), CqlType::Text, true)
.with_field("age".to_string(), CqlType::Int, true),
);
reg
}
fn bare_udt_schema() -> TableSchema {
TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "id".to_string(),
data_type: "int".to_string(),
nullable: false,
default: None,
is_static: false,
},
Column {
name: "addr".to_string(),
data_type: "person".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: std::collections::HashMap::new(),
dropped_columns: std::collections::HashMap::new(),
}
}
const PERSON_USERTYPE: &str = "org.apache.cassandra.db.marshal.UserType(test_ks,706572736f6e,6e616d65:org.apache.cassandra.db.marshal.UTF8Type,616765:org.apache.cassandra.db.marshal.Int32Type)";
fn header_has(stats_path: &std::path::Path, needle: &str) -> bool {
let bytes = std::fs::read(stats_path).expect("read Statistics.db");
bytes.windows(needle.len()).any(|w| w == needle.as_bytes())
}
#[tokio::test]
async fn bare_udt_column_survives_compaction_with_registry() {
let schema = bare_udt_schema();
let registry = person_registry();
let in_dir = tempfile::TempDir::new().expect("in dir");
let mut writer =
crate::storage::sstable::writer::SSTableWriter::with_expected_partitions_and_registry(
in_dir.path().to_path_buf(),
1,
&schema,
1,
Some(®istry),
)
.expect("input writer");
let udt = Value::Udt(Box::new(UdtValue {
type_name: "person".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![
UdtField {
name: "name".to_string(),
value: Some(Value::text("Alice".to_string())),
},
UdtField {
name: "age".to_string(),
value: Some(Value::Integer(30)),
},
],
}));
let mutation = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(1)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: udt,
}],
1_000_000,
None,
);
let key = mutation.decorated_key(&schema).expect("decorated key");
writer
.write_partition(key, vec![mutation])
.expect("write partition");
let input = writer.finish().await.expect("finish input");
assert!(
header_has(&input.stats_path, PERSON_USERTYPE),
"input header must advertise the UDT column as UserType"
);
let out_dir = tempfile::TempDir::new().expect("out dir");
let report = compact_sstables(
vec![input.data_path.clone()],
out_dir.path(),
&schema,
2,
None,
None,
true,
)
.await
.expect("compaction");
assert!(
header_has(&report.output.stats_path, PERSON_USERTYPE),
"compaction output header must keep the UDT column as UserType (not degrade to BytesType)"
);
let mut read_schema = schema.clone();
crate::storage::sstable::writer::data_writer::normalize_schema_udts(
&mut read_schema,
®istry,
);
let config = crate::Config::default();
let platform = std::sync::Arc::new(
crate::platform::Platform::new(&config)
.await
.expect("platform"),
);
let reader = crate::storage::sstable::reader::SSTableReader::open(
&report.output.data_path,
&config,
platform,
)
.await
.expect("open compacted output");
let rows = reader
.iterate_all_partitions_for_compaction(Some(&read_schema))
.await
.expect("compaction iterator");
let mut saw_addr = false;
for row in &rows {
if let crate::storage::sstable::reader::compaction_row::CompactionRowData::Live {
complex,
..
} = &row.row_data
{
if let Some(addr) = complex.iter().find(|c| c.column == "addr") {
saw_addr = true;
assert_eq!(
addr.elements.len(),
2,
"both UDT field cells must survive, got: {:?}",
addr.elements
);
let values: Vec<Option<Value>> =
addr.elements.iter().map(|e| e.value.clone()).collect();
let name_ok = values.iter().any(|v| {
matches!(v, Some(Value::Text(s)) if s == "Alice")
|| matches!(v, Some(Value::Blob(b)) if b.as_ref() == b"Alice")
});
assert!(
name_ok,
"UDT field `name` value must survive compaction, got: {values:?}"
);
let age_ok = values.iter().any(|v| {
matches!(v, Some(Value::Integer(30)))
|| matches!(v, Some(Value::Blob(b)) if b.as_ref() == 30i32.to_be_bytes())
});
assert!(
age_ok,
"UDT field `age` value must survive compaction, got: {values:?}"
);
}
}
}
assert!(
saw_addr,
"compacted output must contain the `addr` UDT as a COMPLEX column (per-field cells \
survived rather than being dropped/collapsed)"
);
}
#[tokio::test]
async fn simple_cell_input_not_upgraded_on_compaction() {
let schema = bare_udt_schema();
let in_dir = tempfile::TempDir::new().expect("in dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::with_expected_partitions(
in_dir.path().to_path_buf(),
1,
&schema,
1,
)
.expect("input writer");
let udt = Value::Udt(Box::new(UdtValue {
type_name: "person".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "name".to_string(),
value: Some(Value::text("Bob".to_string())),
}],
}));
let mutation = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(2)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: udt,
}],
1_000_000,
None,
);
let key = mutation.decorated_key(&schema).expect("decorated key");
writer
.write_partition(key, vec![mutation])
.expect("write partition");
let input = writer.finish().await.expect("finish input");
assert!(
!header_has(&input.stats_path, "UserType("),
"input written without registry must store the column as a simple (non-UserType) cell"
);
let plan = udt_columns_eligible_for_normalization(&[input.data_path.clone()]);
assert!(
plan.eligible_marshals.is_empty() && plan.conflicts.is_empty(),
"a simple-cell input must not be reported as eligible or conflicting"
);
let out_dir = tempfile::TempDir::new().expect("out dir");
let report = compact_sstables(
vec![input.data_path.clone()],
out_dir.path(),
&schema,
2,
None,
None,
true,
)
.await
.expect("compaction must not panic on a simple-cell input");
assert!(
!header_has(&report.output.stats_path, "UserType("),
"compaction must not upgrade a simple-cell input's column to UserType"
);
}
#[tokio::test]
async fn absent_column_in_older_input_does_not_veto_normalization() {
let registry = person_registry();
let full_schema = bare_udt_schema();
let a_dir = tempfile::TempDir::new().expect("a dir");
let mut wa =
crate::storage::sstable::writer::SSTableWriter::with_expected_partitions_and_registry(
a_dir.path().to_path_buf(),
1,
&full_schema,
1,
Some(®istry),
)
.expect("writer a");
let udt = Value::Udt(Box::new(UdtValue {
type_name: "person".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "name".to_string(),
value: Some(Value::text("Carol".to_string())),
}],
}));
let ma = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(1)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: udt,
}],
1_000_000,
None,
);
let ka = ma.decorated_key(&full_schema).expect("key a");
wa.write_partition(ka, vec![ma]).expect("write a");
let input_a = wa.finish().await.expect("finish a");
let older_schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "id".to_string(),
data_type: "int".to_string(),
nullable: false,
default: None,
is_static: false,
},
Column {
name: "note".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: std::collections::HashMap::new(),
dropped_columns: std::collections::HashMap::new(),
};
let b_dir = tempfile::TempDir::new().expect("b dir");
let mut wb = crate::storage::sstable::writer::SSTableWriter::with_expected_partitions(
b_dir.path().to_path_buf(),
1,
&older_schema,
1,
)
.expect("writer b");
let mb = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(2)),
None,
vec![CellOperation::Write {
column: "note".to_string(),
value: Value::text("x".to_string()),
}],
1_000_000,
None,
);
let kb = mb.decorated_key(&older_schema).expect("key b");
wb.write_partition(kb, vec![mb]).expect("write b");
let input_b = wb.finish().await.expect("finish b");
let plan = udt_columns_eligible_for_normalization(&[
input_a.data_path.clone(),
input_b.data_path.clone(),
]);
assert!(
plan.eligible_marshals.contains_key("addr") && plan.conflicts.is_empty(),
"a column absent from an older input must remain eligible, got: {plan:?}"
);
let out_dir = tempfile::TempDir::new().expect("out dir");
let report = compact_sstables(
vec![input_a.data_path.clone(), input_b.data_path.clone()],
out_dir.path(),
&full_schema,
2,
None,
None,
true,
)
.await
.expect("compaction");
assert!(
header_has(&report.output.stats_path, PERSON_USERTYPE),
"newer input's complex UDT column must survive compaction with an older input \
that lacks the column"
);
}
#[tokio::test]
async fn mixed_encoding_inputs_fail_compaction() {
let registry = person_registry();
let schema = bare_udt_schema();
let a_dir = tempfile::TempDir::new().expect("a dir");
let mut wa =
crate::storage::sstable::writer::SSTableWriter::with_expected_partitions_and_registry(
a_dir.path().to_path_buf(),
1,
&schema,
1,
Some(®istry),
)
.expect("writer a");
let udt = Value::Udt(Box::new(UdtValue {
type_name: "person".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "name".to_string(),
value: Some(Value::text("Dave".to_string())),
}],
}));
let ma = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(1)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: udt,
}],
1_000_000,
None,
);
let ka = ma.decorated_key(&schema).expect("key a");
wa.write_partition(ka, vec![ma]).expect("write a");
let input_a = wa.finish().await.expect("finish a");
let b_dir = tempfile::TempDir::new().expect("b dir");
let mut wb = crate::storage::sstable::writer::SSTableWriter::with_expected_partitions(
b_dir.path().to_path_buf(),
1,
&schema,
1,
)
.expect("writer b");
let udt_b = Value::Udt(Box::new(UdtValue {
type_name: "person".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "name".to_string(),
value: Some(Value::text("Eve".to_string())),
}],
}));
let mb = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(2)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: udt_b,
}],
1_000_000,
None,
);
let kb = mb.decorated_key(&schema).expect("key b");
wb.write_partition(kb, vec![mb]).expect("write b");
let input_b = wb.finish().await.expect("finish b");
let plan = udt_columns_eligible_for_normalization(&[
input_a.data_path.clone(),
input_b.data_path.clone(),
]);
assert!(
plan.conflicts.contains("addr"),
"mixed-encoding column must be reported as a conflict, got: {plan:?}"
);
let out_dir = tempfile::TempDir::new().expect("out dir");
let err = compact_sstables(
vec![input_a.data_path.clone(), input_b.data_path.clone()],
out_dir.path(),
&schema,
2,
None,
None,
true,
)
.await
.expect_err("mixed-encoding compaction must fail");
let msg = format!("{err}");
assert!(
msg.contains("disagree on the encoding") && msg.contains("addr"),
"error must explain the mixed-encoding conflict, got: {msg}"
);
}
#[tokio::test]
async fn nested_usertype_column_survives_compaction() {
const INNER_MARSHAL: &str =
"org.apache.cassandra.db.marshal.UserType(test_ks,696e6e6572,78:org.apache.cassandra.db.marshal.Int32Type)";
let outer_marshal = format!(
"org.apache.cassandra.db.marshal.UserType(test_ks,6f75746572,69:{INNER_MARSHAL})"
);
let schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "id".to_string(),
data_type: "int".to_string(),
nullable: false,
default: None,
is_static: false,
},
Column {
name: "addr".to_string(),
data_type: outer_marshal.clone(),
nullable: true,
default: None,
is_static: false,
},
],
comments: std::collections::HashMap::new(),
dropped_columns: std::collections::HashMap::new(),
};
let in_dir = tempfile::TempDir::new().expect("in dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::new(
in_dir.path().to_path_buf(),
1,
&schema,
)
.expect("input writer");
let addr = Value::Udt(Box::new(UdtValue {
type_name: "outer".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "i".to_string(),
value: Some(Value::Udt(Box::new(UdtValue {
type_name: "inner".to_string(),
keyspace: "test_ks".to_string(),
fields: vec![UdtField {
name: "x".to_string(),
value: Some(Value::Integer(5)),
}],
}))),
}],
}));
let mutation = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(1)),
None,
vec![CellOperation::Write {
column: "addr".to_string(),
value: addr,
}],
1_000_000,
None,
);
let key = mutation.decorated_key(&schema).expect("decorated key");
writer.write_partition(key, vec![mutation]).expect("write");
let input = writer.finish().await.expect("finish");
assert!(
header_has(&input.stats_path, &outer_marshal),
"input header must carry the exact nested UserType marshal"
);
let plan = udt_columns_eligible_for_normalization(&[input.data_path.clone()]);
assert_eq!(
plan.eligible_marshals.get("addr").map(String::as_str),
Some(outer_marshal.as_str()),
"compaction must copy the exact nested marshal from the input header"
);
let out_dir = tempfile::TempDir::new().expect("out dir");
let report = compact_sstables(
vec![input.data_path.clone()],
out_dir.path(),
&schema,
2,
None,
None,
true,
)
.await
.expect("compaction");
assert!(
header_has(&report.output.stats_path, &outer_marshal),
"nested UserType column must survive compaction byte-exact"
);
}
#[tokio::test]
async fn newly_added_bare_udt_column_normalized_via_registry() {
let registry = person_registry();
let older_schema = TableSchema {
keyspace: "test_ks".to_string(),
table: "test_table".to_string(),
partition_keys: vec![KeyColumn {
name: "id".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![],
columns: vec![
Column {
name: "id".to_string(),
data_type: "int".to_string(),
nullable: false,
default: None,
is_static: false,
},
Column {
name: "note".to_string(),
data_type: "text".to_string(),
nullable: true,
default: None,
is_static: false,
},
],
comments: std::collections::HashMap::new(),
dropped_columns: std::collections::HashMap::new(),
};
let in_dir = tempfile::TempDir::new().expect("in dir");
let mut writer = crate::storage::sstable::writer::SSTableWriter::with_expected_partitions(
in_dir.path().to_path_buf(),
1,
&older_schema,
1,
)
.expect("writer");
let m = Mutation::new(
TableId::new("test_ks", "test_table"),
PartitionKey::single("id", Value::Integer(1)),
None,
vec![CellOperation::Write {
column: "note".to_string(),
value: Value::text("x".to_string()),
}],
1_000_000,
None,
);
let k = m.decorated_key(&older_schema).expect("key");
writer.write_partition(k, vec![m]).expect("write");
let input = writer.finish().await.expect("finish");
let mut new_schema = bare_udt_schema();
crate::storage::write_engine::merge::apply_udt_marshals_from_inputs(
&mut new_schema,
&[input.data_path.clone()],
Some(®istry),
)
.expect("apply");
let addr = new_schema
.columns
.iter()
.find(|c| c.name == "addr")
.expect("addr column");
assert_eq!(
addr.data_type, PERSON_USERTYPE,
"a UDT column absent from all inputs must be registry-normalized, not left bare"
);
let mut bare_again = bare_udt_schema();
crate::storage::write_engine::merge::apply_udt_marshals_from_inputs(
&mut bare_again,
&[input.data_path.clone()],
None,
)
.expect("apply no-registry");
assert_eq!(
bare_again
.columns
.iter()
.find(|c| c.name == "addr")
.unwrap()
.data_type,
"person",
"without a registry an absent column stays bare"
);
}
#[tokio::test]
async fn unreadable_header_does_not_trigger_registry_normalization() {
let registry = person_registry();
let dir = tempfile::TempDir::new().expect("dir");
let bogus = dir.path().join("nb-1-big-Data.db");
let plan = udt_columns_eligible_for_normalization(&[bogus.clone()]);
assert!(
!plan.headers_verified,
"an unreadable header must leave the plan unverified"
);
let mut schema = bare_udt_schema();
crate::storage::write_engine::merge::apply_udt_marshals_from_inputs(
&mut schema,
&[bogus],
Some(®istry),
)
.expect("apply");
assert_eq!(
schema
.columns
.iter()
.find(|c| c.name == "addr")
.unwrap()
.data_type,
"person",
"with unverified headers, a bare UDT column must NOT be registry-normalized"
);
}
}
#[cfg(all(test, feature = "write-support"))]
mod issue_959_range_tombstone_fixes {
use super::*;
use crate::schema::{Column, KeyColumn};
use crate::storage::write_engine::mutation::ClusteringBound;
use crate::types::Value;
use std::collections::HashMap;
fn dk(byte: u8) -> DecoratedKey {
DecoratedKey::from_key_bytes(vec![byte]).expect("token")
}
fn schema_int_ck(order: crate::schema::ClusteringOrder) -> TableSchema {
TableSchema {
keyspace: "ks".to_string(),
table: "tbl".to_string(),
partition_keys: vec![KeyColumn {
name: "pk".to_string(),
data_type: "int".to_string(),
position: 0,
}],
clustering_keys: vec![crate::schema::ClusteringColumn {
name: "ck".to_string(),
data_type: "int".to_string(),
position: 0,
order,
}],
columns: vec![Column {
name: "v".to_string(),
data_type: "int".to_string(),
nullable: true,
default: None,
is_static: false,
}],
comments: HashMap::new(),
dropped_columns: HashMap::new(),
}
}
fn ck(n: i32) -> ClusteringKey {
ClusteringKey {
columns: vec![("ck".to_string(), Value::Integer(n))],
}
}
fn rt(start: ClusteringBound, end: ClusteringBound, dt: i64) -> RangeTombstone {
RangeTombstone {
start,
end,
deletion_time: dt,
local_deletion_time: (dt / 1_000_000) as i32,
}
}
#[test]
fn overlapping_ranges_coalesce_to_disjoint_segments() {
let schema = schema_int_ck(Default::default());
let mut rts = vec![
(
dk(1),
rt(
ClusteringBound::Inclusive(ck(1)),
ClusteringBound::Inclusive(ck(5)),
100,
),
),
(
dk(1),
rt(
ClusteringBound::Inclusive(ck(2)),
ClusteringBound::Inclusive(ck(3)),
200,
),
),
];
KWayMerger::coalesce_range_tombstones(&mut rts, &schema);
assert_eq!(rts.len(), 3, "union splits into 3 disjoint segments");
assert_eq!(rts[0].1.start, ClusteringBound::Inclusive(ck(1)));
assert_eq!(rts[0].1.end, ClusteringBound::Exclusive(ck(2)));
assert_eq!(rts[0].1.deletion_time, 100);
assert_eq!(rts[1].1.start, ClusteringBound::Inclusive(ck(2)));
assert_eq!(rts[1].1.end, ClusteringBound::Inclusive(ck(3)));
assert_eq!(rts[1].1.deletion_time, 200);
assert_eq!(rts[2].1.start, ClusteringBound::Exclusive(ck(3)));
assert_eq!(rts[2].1.end, ClusteringBound::Inclusive(ck(5)));
assert_eq!(rts[2].1.deletion_time, 100);
for i in 0..rts.len() {
let s = KWayMerger::range_start_cut(&rts[i].1.start);
let e = KWayMerger::range_end_cut(&rts[i].1.end);
assert_ne!(
KWayMerger::cut_cmp(&s, &e, &schema),
Ordering::Greater,
"segment {i} start must not exceed its end"
);
if i > 0 {
let prev_e = KWayMerger::range_end_cut(&rts[i - 1].1.end);
assert_ne!(
KWayMerger::cut_cmp(&prev_e, &s, &schema),
Ordering::Greater,
"segment {i} must not overlap the previous segment"
);
}
}
}
#[test]
fn identical_bounds_collapse_to_newest() {
let schema = schema_int_ck(Default::default());
let mut rts = vec![
(
dk(1),
rt(
ClusteringBound::Inclusive(ck(1)),
ClusteringBound::Inclusive(ck(3)),
100,
),
),
(
dk(1),
rt(
ClusteringBound::Inclusive(ck(1)),
ClusteringBound::Inclusive(ck(3)),
200,
),
),
];
KWayMerger::coalesce_range_tombstones(&mut rts, &schema);
assert_eq!(rts.len(), 1);
assert_eq!(rts[0].1.deletion_time, 200);
assert_eq!(rts[0].1.start, ClusteringBound::Inclusive(ck(1)));
assert_eq!(rts[0].1.end, ClusteringBound::Inclusive(ck(3)));
}
#[test]
fn adjacent_same_deletion_ranges_merge() {
let schema = schema_int_ck(Default::default());
let mut rts = vec![
(
dk(1),
rt(
ClusteringBound::Inclusive(ck(1)),
ClusteringBound::Inclusive(ck(3)),
100,
),
),
(
dk(1),
rt(
ClusteringBound::Exclusive(ck(3)),
ClusteringBound::Inclusive(ck(5)),
100,
),
),
];
KWayMerger::coalesce_range_tombstones(&mut rts, &schema);
assert_eq!(rts.len(), 1, "touching equal-deletion ranges coalesce");
assert_eq!(rts[0].1.start, ClusteringBound::Inclusive(ck(1)));
assert_eq!(rts[0].1.end, ClusteringBound::Inclusive(ck(5)));
}
#[test]
fn distinct_partitions_not_merged() {
let schema = schema_int_ck(Default::default());
let mut rts = vec![
(
dk(1),
rt(ClusteringBound::Bottom, ClusteringBound::Top, 100),
),
(
dk(2),
rt(ClusteringBound::Bottom, ClusteringBound::Top, 100),
),
];
KWayMerger::coalesce_range_tombstones(&mut rts, &schema);
assert_eq!(rts.len(), 2);
assert_ne!(rts[0].0.key, rts[1].0.key);
}
#[test]
fn newer_complex_deletion_survives_full_range_shadow() {
let schema = schema_int_ck(Default::default());
let entry = MergeEntry::new(
0,
dk(1),
Some(ck(2)),
50,
RowData::Live {
cells: vec![CellData::new("ck".to_string(), Value::Integer(2), 50)],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "v".to_string(),
marked_for_delete_at: 200,
local_deletion_time: 0,
}]);
let range = vec![(
dk(1),
rt(ClusteringBound::Bottom, ClusteringBound::Top, 100),
)];
let out = KWayMerger::apply_range_shadowing(entry, &range, &schema)
.expect("a complex deletion newer than the range must survive");
assert_eq!(
out.complex_deletions.len(),
1,
"the newer collection deletion is preserved"
);
assert_eq!(out.complex_deletions[0].marked_for_delete_at, 200);
match out.row_data {
RowData::Live { cells } => assert!(cells.is_empty(), "carrier holds no data cells"),
other => panic!("expected an empty Live carrier, got {other:?}"),
}
}
#[test]
fn older_complex_deletion_subsumed_by_range() {
let schema = schema_int_ck(Default::default());
let entry = MergeEntry::new(
0,
dk(1),
Some(ck(2)),
50,
RowData::Live {
cells: vec![CellData::new("ck".to_string(), Value::Integer(2), 50)],
},
)
.with_complex_deletions(vec![ComplexDeletion {
column: "v".to_string(),
marked_for_delete_at: 50,
local_deletion_time: 0,
}]);
let range = vec![(
dk(1),
rt(ClusteringBound::Bottom, ClusteringBound::Top, 100),
)];
assert!(
KWayMerger::apply_range_shadowing(entry, &range, &schema).is_none(),
"an older complex deletion is subsumed; nothing survives"
);
}
}