#![allow(
clippy::doc_markdown,
clippy::default_trait_access,
reason = "test code"
)]
#![expect(clippy::expect_used, reason = "test code")]
use super::*;
use crate::{config::BloomConstructionPolicy, fs::StdFs, hash::hash64};
use tempfile::tempdir;
use test_log::test;
fn test_recover_params(file_path: PathBuf, checksum: Checksum) -> RecoverParams {
let mut params = RecoverParams::new(
file_path,
checksum,
0,
Arc::new(StdFs),
crate::comparator::default_comparator(),
Arc::new(Cache::with_capacity_bytes(1_000_000)),
);
params.descriptor_table = Some(Arc::new(DescriptorTable::new(10)));
params
}
fn test_with_table(
items: &[InternalValue],
f: impl Fn(Table) -> crate::Result<()>,
rotate_every: Option<usize>,
config_writer: Option<impl Fn(Writer) -> Writer>,
) -> crate::Result<()> {
test_with_table_impl(
items,
f,
rotate_every,
config_writer,
#[cfg(zstd_any)]
None,
)
}
#[expect(
clippy::too_many_lines,
clippy::cognitive_complexity,
clippy::cast_possible_truncation,
clippy::unwrap_used
)]
fn test_with_table_impl(
items: &[InternalValue],
f: impl Fn(Table) -> crate::Result<()>,
rotate_every: Option<usize>,
config_writer: Option<impl Fn(Writer) -> Writer>,
#[cfg(zstd_any)] zstd_dictionary: Option<Arc<crate::compression::ZstdDictionary>>,
) -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
{
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
#[cfg(zstd_any)]
if zstd_dictionary.is_some() {
writer = writer.use_zstd_dictionary(zstd_dictionary.clone());
}
if let Some(f) = &config_writer {
writer = f(writer);
}
for (idx, item) in items.iter().enumerate() {
if let Some(rotate) = rotate_every
&& idx % rotate == 0
{
writer.spill_block()?;
}
writer.write(item.clone())?;
}
let (_, checksum) = writer.finish()?.unwrap();
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
#[cfg_attr(not(any(zstd_any, feature = "metrics")), expect(unused_mut))]
let mut params = test_recover_params(file.clone(), checksum);
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_none(), "should use full index");
assert_eq!(0, table.pinned_block_index_size(), "should not pin index");
assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_filter = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_none(), "should use full index");
assert_eq!(0, table.pinned_block_index_size(), "should not pin index");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_none(), "should use full index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_filter = true;
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_none(), "should use full index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.descriptor_table = None;
params.pin_filter = true;
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_none(), "should use full index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert!(matches!(table.file_accessor, FileAccessor::File(..)));
f(table)?;
}
}
std::fs::remove_file(&file)?;
{
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_partitioned_index();
#[cfg(zstd_any)]
if zstd_dictionary.is_some() {
writer = writer.use_zstd_dictionary(zstd_dictionary.clone());
}
if let Some(f) = config_writer {
writer = f(writer);
}
for (idx, item) in items.iter().enumerate() {
if let Some(rotate) = rotate_every
&& idx % rotate == 0
{
writer.spill_block()?;
}
writer.write(item.clone())?;
}
let (_, checksum) = writer.finish()?.unwrap();
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
#[cfg_attr(not(any(zstd_any, feature = "metrics")), expect(unused_mut))]
let mut params = test_recover_params(file.clone(), checksum);
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_some(), "should use two-level index");
assert_eq!(0, table.pinned_filter_size(), "should not pin filter");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_filter = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_some(), "should use two-level index");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_some(), "should use two-level index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.pin_filter = true;
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.clone()
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_some(), "should use two-level index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert!(matches!(
table.file_accessor,
FileAccessor::DescriptorTable { .. }
));
f(table)?;
}
{
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file, checksum);
params.descriptor_table = None;
params.pin_filter = true;
params.pin_index = true;
#[cfg(zstd_any)]
{
params.zstd_dictionaries = zstd_dictionary
.map_or_else(crate::compression::ZstdDictionaries::new, |dict| {
crate::compression::ZstdDictionaries::new().with(dict)
});
}
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
assert_eq!(0, table.id());
assert_eq!(items.len(), table.metadata.item_count as usize);
assert!(table.regions.index.is_some(), "should use two-level index");
assert!(table.pinned_block_index_size() > 0, "should pin index");
assert!(matches!(table.file_accessor, FileAccessor::File(..)));
f(table)?;
}
}
Ok(())
}
#[cfg(feature = "zstd")]
fn test_with_table_and_zstd_dictionary(
items: &[InternalValue],
f: impl Fn(Table) -> crate::Result<()>,
rotate_every: Option<usize>,
config_writer: Option<impl Fn(Writer) -> Writer>,
zstd_dictionary: Arc<crate::compression::ZstdDictionary>,
) -> crate::Result<()> {
test_with_table_impl(items, f, rotate_every, config_writer, Some(zstd_dictionary))
}
#[cfg(feature = "zstd")]
fn make_test_dictionary() -> crate::compression::ZstdDictionary {
let mut samples = Vec::new();
for i in 0u32..500 {
let key = format!("key-{i:05}");
let val = format!("value-{i:05}-padding-to-make-it-longer");
samples.extend_from_slice(key.as_bytes());
samples.extend_from_slice(val.as_bytes());
}
crate::compression::ZstdDictionary::new(&samples)
}
#[cfg(feature = "zstd")]
#[test]
#[expect(clippy::unwrap_used)]
fn block_layout_section_roundtrips_for_large_zstd_blocks() {
use crate::cache::Cache;
use crate::fs::StdFs;
#[cfg(feature = "metrics")]
use crate::metrics::Metrics;
use crate::table::Writer;
let items: Vec<crate::InternalValue> = (0u64..20_000)
.map(|i| {
crate::InternalValue::from_components(
format!("key-{i:012}").into_bytes(),
format!("value-{i:08}-payload").into_bytes(),
1,
crate::ValueType::Value,
)
})
.collect();
let dir = tempdir().unwrap();
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
.unwrap()
.use_data_block_size(256 * 1024)
.use_data_block_compression(crate::CompressionType::Zstd(19));
for item in &items {
writer.write(item.clone()).unwrap();
}
let (_, checksum) = writer.finish().unwrap().unwrap();
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(file, checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(4_000_000));
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params).unwrap()
};
assert!(
table.regions.block_layout.is_some(),
"large multi-inner-block table must carry a block_layout section",
);
assert!(
!table.block_layout.is_empty(),
"at least one data block must have a recorded inner-block layout",
);
for offset in table.block_layout.offsets() {
let ends = table
.block_layout
.ends_for(offset)
.expect("offsets() entries must resolve via ends_for");
assert!(
ends.len() >= 2,
"recorded block must have >= 2 inner blocks"
);
assert!(
ends.windows(2).all(|w| w[0] < w[1]),
"cumulative ends must be strictly increasing: {ends:?}",
);
}
let small_file = dir.path().join("table-small");
let mut small_writer = Writer::new(small_file.clone(), 0, 0, Arc::new(StdFs))
.unwrap()
.use_data_block_size(4 * 1024)
.use_data_block_compression(crate::CompressionType::Zstd(19));
for item in &items {
small_writer.write(item.clone()).unwrap();
}
let (_, small_checksum) = small_writer.finish().unwrap().unwrap();
#[cfg(feature = "metrics")]
let small_metrics = Arc::new(Metrics::default());
let small_table = {
let mut params = test_recover_params(small_file, small_checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(4_000_000));
#[cfg(feature = "metrics")]
{
params.metrics = small_metrics;
}
Table::recover(params).unwrap()
};
assert!(
small_table.regions.block_layout.is_none(),
"default small-block table must NOT carry a block_layout section",
);
assert_eq!(
small_table.block_layout.len(),
0,
"small-block table's layout map must be empty",
);
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read() -> crate::Result<()> {
let items = [crate::InternalValue::from_components(
b"abc",
b"asdasdasd",
3,
crate::ValueType::Value,
)];
test_with_table(
&items,
|table| {
assert_eq!(
b"abc",
&*table
.get(b"abc", SeqNo::MAX, hash64(b"abc"))?
.unwrap()
.key
.user_key,
);
assert_eq!(None, table.get(b"def", SeqNo::MAX, hash64(b"def"))?,);
assert_eq!(None, table.get(b"____", SeqNo::MAX, hash64(b"____"))?,);
assert_eq!(
table.metadata.key_range,
crate::KeyRange::new((b"abc".into(), b"abc".into())),
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn restricted_view_clamps_point_and_range_reads() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let restricted = table.with_restriction(crate::UserKey::from(&b"c"[..]));
assert_eq!(None, restricted.get(b"a", SeqNo::MAX, hash64(b"a"))?);
assert_eq!(None, restricted.get(b"b", SeqNo::MAX, hash64(b"b"))?);
assert!(restricted.get(b"c", SeqNo::MAX, hash64(b"c"))?.is_some());
assert!(restricted.get(b"d", SeqNo::MAX, hash64(b"d"))?.is_some());
assert!(table.get(b"a", SeqNo::MAX, hash64(b"a"))?.is_some());
let keys: Vec<_> = restricted
.range(..)
.map(|r| r.unwrap().key.user_key)
.collect();
assert_eq!(
keys,
vec![
crate::UserKey::from(&b"c"[..]),
crate::UserKey::from(&b"d"[..]),
crate::UserKey::from(&b"e"[..]),
],
);
let cmp = crate::comparator::default_comparator();
assert!(!restricted.check_key_range_overlap_cmp(
&(
core::ops::Bound::Unbounded,
core::ops::Bound::Excluded(&b"c"[..]),
),
cmp.as_ref(),
));
assert!(restricted.check_key_range_overlap_cmp(
&(
core::ops::Bound::Included(&b"d"[..]),
core::ops::Bound::Unbounded,
),
cmp.as_ref(),
));
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn reopen_restricted_yields_a_distinct_clamped_view() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let restricted = table.reopen_restricted(crate::UserKey::from(&b"c"[..]))?;
assert_eq!(None, restricted.get(b"a", SeqNo::MAX, hash64(b"a"))?);
assert_eq!(None, restricted.get(b"b", SeqNo::MAX, hash64(b"b"))?);
assert!(restricted.get(b"c", SeqNo::MAX, hash64(b"c"))?.is_some());
assert!(restricted.get(b"e", SeqNo::MAX, hash64(b"e"))?.is_some());
assert!(table.get(b"a", SeqNo::MAX, hash64(b"a"))?.is_some());
let keys: Vec<_> = restricted
.range(..)
.map(|r| r.unwrap().key.user_key)
.collect();
assert_eq!(
keys,
vec![
crate::UserKey::from(&b"c"[..]),
crate::UserKey::from(&b"d"[..]),
crate::UserKey::from(&b"e"[..]),
],
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
#[cfg(feature = "columnar")]
fn metadata_bounds_accept_a_legacy_bitmap_when_digest_authenticated() -> crate::Result<()> {
use crate::config::DeleteStrategy;
let dir = tempdir()?;
let file = dir.path().join("legacy");
let fs: Arc<dyn Fs> = Arc::new(StdFs);
let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?
.use_columnar(true)
.use_zone_map(true)
.delete_strategy(DeleteStrategy::MergeOnRead);
writer.omit_delete_bitmap_hash_for_test = true;
for i in 0..16u32 {
writer.write(crate::InternalValue::from_components(
format!("key{i:03}").into_bytes(),
b"v".as_slice(),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
writer.delete_bitmap_mut().insert(3);
assert!(writer.finish()?.is_some(), "legacy SST is non-empty");
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
assert!(
matches!(
table.verify_reconcile_gates(None, false),
Err((crate::table::ReconcileGate::MetadataBounds, _))
),
"repair (no matching digest) keeps failing closed on the \
unauthenticatable legacy bitmap",
);
if let Err((gate, e)) = table.verify_reconcile_gates(None, true) {
panic!("an authenticated digest accepts the legacy bitmap, {gate:?} refused it: {e}");
}
Ok(())
}
#[test]
#[expect(clippy::unwrap_used)]
fn restricted_view_scan_starts_at_the_bound() -> crate::Result<()> {
let items: Vec<_> = (0..40u32)
.map(|i| {
crate::InternalValue::from_components(
format!("key{i:03}").into_bytes(),
b"v".as_slice(),
0,
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let restricted = table.reopen_restricted(crate::UserKey::from(&b"key020"[..]))?;
let keys: Vec<_> = restricted
.scan()?
.map(|r| r.unwrap().key.user_key)
.collect();
let expected: Vec<_> = (20..40u32)
.map(|i| crate::UserKey::from(format!("key{i:03}").into_bytes()))
.collect();
assert_eq!(
keys, expected,
"the restricted scan must yield only keys at or past the bound",
);
let full = table.scan()?.count();
assert_eq!(full, 40, "the unrestricted view scans the whole file");
Ok(())
},
None,
Some(|w: Writer| w.use_data_block_size(64)),
)
}
#[test]
fn restricted_view_clamps_visible_range_tombstones() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"z", b"v", 0, crate::ValueType::Value),
];
let rt = |s: &[u8], e: &[u8], seqno| {
crate::range_tombstone::RangeTombstone::new(
crate::UserKey::from(s),
crate::UserKey::from(e),
seqno,
)
};
test_with_table(
&items,
|table| {
let unrestricted: Vec<_> = table.visible_range_tombstones().collect();
assert_eq!(
unrestricted.len(),
3,
"the unrestricted view keeps the full list: {unrestricted:?}",
);
let restricted = table.reopen_restricted(crate::UserKey::from(&b"g"[..]))?;
let visible: Vec<_> = restricted
.visible_range_tombstones()
.map(alloc::borrow::Cow::into_owned)
.collect();
assert_eq!(
visible,
vec![rt(b"g", b"m", 5), rt(b"p", b"r", 6)],
"wholly-below dropped, straddling clamped to the bound, \
above-bound untouched",
);
Ok(())
},
None,
Some(|mut w: Writer| {
w.write_range_tombstone(rt(b"a", b"c", 4)); w.write_range_tombstone(rt(b"a", b"m", 5)); w.write_range_tombstone(rt(b"p", b"r", 6)); w
}),
)
}
#[test]
fn blob_links_do_not_cache_the_tables_descriptor() -> crate::Result<()> {
use crate::AbstractTree;
let dir = tempdir()?;
let descriptors = Arc::new(crate::DescriptorTable::new(16));
let tree = crate::Config::new(
dir.path(),
crate::SequenceNumberCounter::default(),
crate::SequenceNumberCounter::default(),
)
.use_descriptor_table(Some(descriptors.clone()))
.with_kv_separation(Some(crate::KvSeparationOptions::default()))
.open()?;
for i in 0..10u64 {
tree.insert(format!("key{i:04}"), vec![b'v'; 2_048], i);
}
tree.flush_active_memtable(0)?;
let version = tree.current_version();
let Some(table) = version.iter_tables().next() else {
panic!("the flush wrote a table");
};
descriptors.remove_for_table(&table.global_id());
assert_eq!(table.blob_links()?.len(), 1, "one blob file is referenced");
assert!(
descriptors.peek_for_table(&table.global_id()).is_none(),
"the link read must not cache the descriptor",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn verify_blob_links_rejects_a_narrowed_key_span() -> crate::Result<()> {
use crate::blob_tree::handle::BlobIndirection;
use crate::coding::Encode;
use crate::table::Writer;
use crate::vlog::ValueHandle;
use crate::{InternalValue, ValueType};
let dir = tempdir()?;
let write = |name: &str, last_key: &str| -> crate::Result<Table> {
let file = dir.path().join(name);
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..10 {
let value = BlobIndirection {
size: 1000,
vhandle: ValueHandle {
blob_file_id: 0,
on_disk_size: 500,
offset: i * 500,
},
}
.encode_into_vec();
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
value,
i + 1,
ValueType::Indirection,
))?;
}
w.link_blob_file(crate::table::writer::LinkedFile {
blob_file_id: 0,
len: 10,
bytes: 10_000,
on_disk_bytes: 5_000,
first_key: b"key00000".into(),
last_key: last_key.as_bytes().into(),
});
w.finish()?.expect("the SST is non-empty").1
};
Table::recover(test_recover_params(file, checksum))
};
let suffix = || crate::UserKey::from(&b"key00005"[..]);
let honest = write("honest", "key00009")?;
honest.verify_blob_links()?;
honest.reopen_restricted(suffix())?.verify_blob_links()?;
let narrowed = write("narrowed", "key00007")?;
assert!(
narrowed.verify_blob_links().is_err(),
"the exact check must reject a span that stops short of the last key",
);
assert!(
narrowed
.reopen_restricted(suffix())?
.verify_blob_links()
.is_err(),
"the containment check must reject a span the suffix reaches past",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn verify_blob_links_rejects_an_undercounted_suffix_id_on_a_restricted_view() -> crate::Result<()> {
use crate::blob_tree::handle::BlobIndirection;
use crate::coding::Encode;
use crate::table::Writer;
use crate::vlog::ValueHandle;
use crate::{InternalValue, ValueType};
let dir = tempdir()?;
let file = dir.path().join("0");
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..10 {
let value = BlobIndirection {
size: 1000,
vhandle: ValueHandle {
blob_file_id: i,
on_disk_size: 500,
offset: 0,
},
}
.encode_into_vec();
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
value,
i + 1,
ValueType::Indirection,
))?;
}
for i in 0u64..10 {
let bytes = if i == 9 { 1 } else { 1000 };
let key = crate::UserKey::from(format!("key{i:05}").into_bytes());
w.link_blob_file(crate::table::writer::LinkedFile {
blob_file_id: i,
len: 1,
bytes,
on_disk_bytes: 500,
first_key: key.clone(),
last_key: key,
});
}
w.finish()?.expect("the SST is non-empty").1
};
let recover =
|| -> crate::Result<Table> { Table::recover(test_recover_params(file.clone(), checksum)) };
assert!(
recover()?.verify_blob_links().is_err(),
"the unrestricted exact check must reject the under-counted id",
);
let restricted = recover()?.reopen_restricted(crate::UserKey::from(&b"key00005"[..]))?;
assert!(
restricted.verify_blob_links().is_err(),
"the restricted containment check must reject a suffix id the section under-counts",
);
Ok(())
}
#[test]
fn a_restricted_view_reopened_with_a_descriptor_cache_keeps_its_file_there() -> crate::Result<()> {
let dir = tempfile::tempdir()?;
let file = dir.path().join("table");
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for i in 0u64..20 {
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"v".to_vec(),
i + 1,
crate::ValueType::Value,
))?;
}
w.finish()?.expect("the SST is non-empty").1
};
let mut params = test_recover_params(file, checksum);
params.descriptor_table = None;
let table = Table::recover(params)?;
let bound = crate::UserKey::from(&b"key00010"[..]);
let key = b"key00015";
let hash = crate::hash::hash64(key);
let plain = table.reopen_restricted(bound.clone())?;
assert!(plain.get(key, crate::SeqNo::MAX, hash)?.is_some());
drop(plain);
let cache = Arc::new(DescriptorTable::new(10));
let held = table.reopen_restricted_with(bound, Some((Arc::clone(&cache), 777)))?;
assert_eq!(held.global_id().tree_id(), 777);
assert!(held.get(key, crate::SeqNo::MAX, hash)?.is_some());
assert_eq!(
cache.len(),
1,
"the view's file is in the cache it was given"
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "std"))]
#[test]
fn verify_blob_links_on_a_restricted_view_counts_and_caches_nothing() -> crate::Result<()> {
use crate::blob_tree::handle::BlobIndirection;
use crate::coding::Encode;
use crate::vlog::ValueHandle;
use crate::{InternalValue, ValueType};
let dir = tempdir()?;
let file = dir.path().join("0");
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..10 {
let value = BlobIndirection {
size: 1000,
vhandle: ValueHandle {
blob_file_id: i,
on_disk_size: 500,
offset: 0,
},
}
.encode_into_vec();
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
value,
i + 1,
ValueType::Indirection,
))?;
let key = crate::UserKey::from(format!("key{i:05}").into_bytes());
w.link_blob_file(crate::table::writer::LinkedFile {
blob_file_id: i,
len: 1,
bytes: 1000,
on_disk_bytes: 500,
first_key: key.clone(),
last_key: key,
});
}
w.finish()?.expect("the SST is non-empty").1
};
let cache = Arc::new(crate::Cache::with_capacity_bytes(1 << 20));
let mut params = test_recover_params(file, checksum);
params.cache = cache.clone();
params.metrics = Arc::new(Metrics::default());
let metrics = params.metrics.clone();
let restricted =
Table::recover(params)?.reopen_restricted(crate::UserKey::from(&b"key00005"[..]))?;
let (read, decoded, copied) = (
metrics.bytes_read(),
metrics.bytes_decoded(),
metrics.bytes_copied(),
);
let cached = cache.size();
restricted.verify_blob_links()?;
assert_eq!(
(
metrics.bytes_read(),
metrics.bytes_decoded(),
metrics.bytes_copied()
),
(read, decoded, copied),
"the verification walk is not a caller's read",
);
assert_eq!(
cache.size(),
cached,
"the verification walk must not fill the block cache",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn verify_blob_links_leaves_the_descriptor_table_as_it_found_it() -> crate::Result<()> {
use crate::blob_tree::handle::BlobIndirection;
use crate::coding::Encode;
use crate::vlog::ValueHandle;
use crate::{InternalValue, ValueType};
let dir = tempdir()?;
let file = dir.path().join("0");
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..10 {
let value = BlobIndirection {
size: 1000,
vhandle: ValueHandle {
blob_file_id: i,
on_disk_size: 500,
offset: 0,
},
}
.encode_into_vec();
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
value,
i + 1,
ValueType::Indirection,
))?;
let key = crate::UserKey::from(format!("key{i:05}").into_bytes());
w.link_blob_file(crate::table::writer::LinkedFile {
blob_file_id: i,
len: 1,
bytes: 1000,
on_disk_bytes: 500,
first_key: key.clone(),
last_key: key,
});
}
w.finish()?.expect("the SST is non-empty").1
};
let descriptors = Arc::new(DescriptorTable::new(10));
let mut params = test_recover_params(file, checksum);
params.descriptor_table = Some(descriptors.clone());
let table = Table::recover(params)?;
table.file_accessor.remove_for_table(&table.global_id());
assert_eq!(descriptors.len(), 0, "precondition: the descriptor is cold");
table.verify_blob_links()?;
assert_eq!(
descriptors.len(),
0,
"verifying the table must not cache its descriptor",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn an_untraced_read_leaves_the_descriptor_table_as_it_found_it() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("0");
let checksum = {
let mut w = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..50 {
w.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value".to_vec(),
i + 1,
crate::ValueType::Value,
))?;
}
w.finish()?.expect("the SST is non-empty").1
};
let descriptors = Arc::new(DescriptorTable::new(10));
let mut params = test_recover_params(file, checksum);
params.descriptor_table = Some(descriptors.clone());
let table = Table::recover(params)?;
table.file_accessor.remove_for_table(&table.global_id());
assert_eq!(descriptors.len(), 0, "precondition: the descriptor is cold");
let mut rows = 0;
for kv in table.range_iter(..).untraced() {
kv?;
rows += 1;
}
assert_eq!(rows, 50, "the untraced walk reads every row");
assert_eq!(
descriptors.len(),
0,
"an untraced read must not cache the descriptor it opened",
);
Ok(())
}
#[cfg(all(feature = "zstd", feature = "std"))]
#[test]
fn an_untraced_bounded_range_leaves_no_partial_block_in_the_cache() -> crate::Result<()> {
use crate::{
AbstractTree, CompressionType, SequenceNumberCounter,
config::{BlockSizePolicy, CompressionPolicy},
};
super::iter::force_partial_decode_on_this_thread();
let dir = tempdir()?;
let crate::AnyTree::Standard(tree) = crate::Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::zstd(19)?))
.data_block_size_policy(BlockSizePolicy::all(512 * 1024))
.open()?
else {
unreachable!("standard tree configured");
};
for i in 0u64..20_000 {
tree.insert(
format!("key-{i:08}"),
format!("value-{i:08}-padding-padding"),
0,
);
}
tree.flush_active_memtable(0)?;
let binding = tree.version_history.read().latest_version();
let Some(table) = binding.version.iter_tables().next() else {
unreachable!("flush produced one table");
};
let first_block = table
.untraced_index_walk()
.next()
.expect("the table has a data block")?
.offset();
let cached = table.cache.size();
let bounded =
|| crate::UserKey::from(&b"key-00000010"[..])..crate::UserKey::from(&b"key-00000030"[..]);
let mut rows = 0;
for kv in table.range_iter(bounded()).untraced() {
kv?;
rows += 1;
}
assert_eq!(rows, 20, "the bounded range reads its rows");
assert_eq!(
table.cache.size(),
cached,
"an untraced read must not cache a partial block",
);
for kv in table.range_iter(bounded()) {
kv?;
}
assert!(
table
.cache
.peek_partial_block(table.global_id(), first_block)
.is_some(),
"precondition: a traced bounded range takes the partial decode path",
);
Ok(())
}
#[cfg(feature = "std")]
#[derive(Clone)]
struct SharedInodeFs(crate::fs::MemFs, Arc<core::sync::atomic::AtomicU64>);
#[cfg(feature = "std")]
impl crate::fs::Fs for SharedInodeFs {
fn open(
&self,
path: &std::path::Path,
options: &crate::fs::FsOpenOptions,
) -> crate::io::Result<Box<dyn crate::fs::FsFile>> {
self.0.open(path, options)
}
fn remove_file(&self, path: &std::path::Path) -> crate::io::Result<()> {
self.0.remove_file(path)
}
fn rename(&self, from: &std::path::Path, to: &std::path::Path) -> crate::io::Result<()> {
self.0.rename(from, to)
}
fn create_dir_all(&self, path: &std::path::Path) -> crate::io::Result<()> {
self.0.create_dir_all(path)
}
fn remove_dir_all(&self, path: &std::path::Path) -> crate::io::Result<()> {
self.0.remove_dir_all(path)
}
fn sync_directory(&self, path: &std::path::Path) -> crate::io::Result<()> {
self.0.sync_directory(path)
}
fn read_dir(&self, path: &std::path::Path) -> crate::io::Result<Vec<crate::fs::FsDirEntry>> {
self.0.read_dir(path)
}
fn metadata(&self, path: &std::path::Path) -> crate::io::Result<crate::fs::FsMetadata> {
self.0.metadata(path)
}
fn exists(&self, path: &std::path::Path) -> crate::io::Result<bool> {
self.0.exists(path)
}
fn capabilities(&self, path: &std::path::Path) -> crate::fs::FsCapabilities {
self.0.capabilities(path)
}
fn punch_hole(&self, path: &std::path::Path, offset: u64, len: u64) -> crate::io::Result<()> {
self.0.punch_hole(path, offset, len)
}
fn hard_link_count(&self, _path: &std::path::Path) -> crate::io::Result<u64> {
Ok(self.1.load(core::sync::atomic::Ordering::Acquire))
}
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_refuses_a_hard_linked_table() -> crate::Result<()> {
use crate::fs::Fs;
let memfs = crate::fs::MemFs::new();
let shared: Arc<dyn Fs> = Arc::new(SharedInodeFs(
memfs.clone(),
Arc::new(core::sync::atomic::AtomicU64::new(2)),
));
let plain: Arc<dyn Fs> = Arc::new(memfs);
let root = std::path::absolute("/db")?;
shared.create_dir_all(&root)?;
let build = |fs: &Arc<dyn Fs>, name: &str| -> crate::Result<(std::path::PathBuf, Checksum)> {
let path = root.join(name);
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(fs))?.use_data_block_size(256);
for i in 0..256u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
Ok((path, checksum))
};
let recover = |fs: &Arc<dyn Fs>, path: &std::path::Path, checksum| -> crate::Result<Table> {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.to_path_buf(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)
};
let read_all = |fs: &Arc<dyn Fs>, path: &std::path::Path| -> crate::Result<Vec<u8>> {
let file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
let len = crate::fs::FsFile::metadata(&*file)?.len;
Ok(crate::file::read_exact(&*file, 0, usize::try_from(len).unwrap_or(0))?.to_vec())
};
let (shared_path, shared_checksum) = build(&shared, "0")?;
let before = read_all(&shared, &shared_path)?;
let table = recover(&shared, &shared_path, shared_checksum)?;
let punch = table.punch_offset_for(b"k0128")?;
assert!(punch > 0, "the fixture has a punchable prefix");
table.mark_punch_on_drop(punch);
drop(table);
assert_eq!(
read_all(&shared, &shared_path)?,
before,
"a hard-linked SST must not be punched: the shared inode is the \
checkpoint's data too",
);
let (plain_path, plain_checksum) = build(&plain, "1")?;
let table = recover(&plain, &plain_path, plain_checksum)?;
let punch = table.punch_offset_for(b"k0128")?;
table.mark_punch_on_drop(punch);
drop(table);
let after = read_all(&plain, &plain_path)?;
assert!(
after
.get(..64)
.is_some_and(|head| head.iter().all(|&b| b == 0)),
"an exclusively-owned SST is still reclaimed",
);
let (paused_path, paused_checksum) = build(&plain, "2")?;
let before = read_all(&plain, &paused_path)?;
let table = recover(&plain, &paused_path, paused_checksum)?;
let pause = crate::deletion_pause::DeletionPause::new_shared();
table.install_deletion_pause(std::sync::Arc::clone(&pause));
let guard = pause.acquire();
let punch = table.punch_offset_for(b"k0128")?;
table.mark_punch_on_drop(punch);
drop(table);
assert_eq!(
read_all(&plain, &paused_path)?,
before,
"an active checkpoint pause must defer the SST prefix reclaim",
);
drop(guard);
let after = read_all(&plain, &paused_path)?;
assert!(
after
.get(..64)
.is_some_and(|head| head.iter().all(|&b| b == 0)),
"releasing the pause must run the deferred reclaim",
);
assert_eq!(
after.len(),
before.len(),
"the deferred reclaim punches, it does not truncate",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punched_run_with_allocated_edges_still_proves_the_punch() -> crate::Result<()> {
use crate::fs::Fs;
use std::io::{Seek, SeekFrom, Write};
let memfs = crate::fs::MemFs::new();
let fs: Arc<dyn Fs> = Arc::new(memfs.clone());
let root = std::path::absolute("/db")?;
fs.create_dir_all(&root)?;
let path = root.join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
for i in 0..256u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let punch_off = table.punch_offset_for(b"k0128")?;
assert!(punch_off > 64, "the fixture has a punchable prefix");
{
let mut file = fs.open(
&path,
&crate::fs::FsOpenOptions::new().read(true).write(true),
)?;
file.seek(SeekFrom::Start(0))?;
file.write_all(&vec![
0u8;
usize::try_from(punch_off).expect("small fixture")
])?;
}
assert!(
matches!(
table.punch_geometry()?.verdict,
crate::table::PunchProbe::Unpunched
),
"zeros without a hole are corruption, never a punch",
);
let mid = punch_off / 2;
memfs.punch_hole(&path, mid - 8, 16)?;
assert!(
matches!(
table.punch_geometry()?.verdict,
crate::table::PunchProbe::Punched
),
"a hole contained in the zeroed run proves the punch even though no \
block's whole extent is one",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn successive_views_punch_only_above_what_the_last_one_finished() -> crate::Result<()> {
use super::inner::PUNCH_ALIGN;
use crate::fs::Fs;
let fault = crate::fs::FaultFs::new(crate::fs::MemFs::new());
let injector = fault.injector();
let fs: Arc<dyn Fs> = Arc::new(fault);
let root = std::path::absolute("/db")?;
fs.create_dir_all(&root)?;
let path = root.join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(4_096);
for i in 0..20_000u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:06}").into_bytes(),
[b'v'; 100],
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let first = {
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
let second = first.reopen_restricted(b"k005000".to_vec().into())?;
let off1 = first.punch_offset_for(b"k005000")?;
first.mark_punch_on_drop(off1);
drop(first);
let third = second.reopen_restricted(b"k010000".to_vec().into())?;
let off2 = second.punch_offset_for(b"k010000")?;
second.mark_punch_on_drop(off2);
drop(second);
drop(third);
let punches = injector.punches_for(&path.to_string_lossy());
let floor = off1 - off1 % PUNCH_ALIGN;
let (second_view, first_view): (Vec<_>, Vec<_>) =
punches.iter().copied().partition(|(o, len)| o + len > off1);
assert!(
second_view.iter().all(|(o, _)| *o >= floor),
"the second view punches only from {floor} up: {punches:?}",
);
let covered = |extents: &[(u64, u64)]| extents.iter().map(|(_, len)| len).sum::<u64>();
assert_eq!(
off1,
covered(&first_view),
"the first view punched its prefix"
);
assert_eq!(
off2 - floor,
covered(&second_view),
"the second view punched the rest, from the aligned edge",
);
Ok(())
}
#[test]
fn punch_extents_cut_at_block_edges_highest_first() {
use super::inner::{PUNCH_CHUNK, punch_extents_top_down};
assert!(punch_extents_top_down(&[]).is_empty(), "no block, no call");
assert_eq!(
punch_extents_top_down(&[(100, 5_000)]),
vec![(100, 4_900)],
"one block is one call",
);
let c = PUNCH_CHUNK;
let third = c / 3;
let blocks: Vec<(u64, u64)> = (0..7).map(|i| (i * third, (i + 1) * third)).collect();
assert_eq!(
punch_extents_top_down(&blocks),
vec![(4 * third, 3 * third), (third, 3 * third), (0, third)],
"adjacent blocks are taken whole, as many as fit in the chunk, top-down",
);
assert_eq!(
punch_extents_top_down(&[(0, 10), (10, 2 * c + 10)]),
vec![(10, 2 * c), (0, 10)],
"a block larger than the chunk is one call of its own",
);
assert_eq!(
punch_extents_top_down(&[(0, 100), (200, 300)]),
vec![(200, 100), (0, 100)],
"blocks apart stay apart, the higher one first",
);
}
#[cfg(all(feature = "std", target_os = "linux", target_pointer_width = "64"))]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_frees_the_prefix_pages_on_a_real_filesystem() -> crate::Result<()> {
use crate::fs::Fs;
let dir = tempfile::tempdir()?;
let fs: Arc<dyn Fs> = Arc::new(crate::fs::StdFs);
if !fs.capabilities(dir.path()).punch_hole {
eprintln!("skipping: the temporary directory's file system cannot punch holes");
return Ok(());
}
let path = dir.path().join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(4_096);
for i in 0..8_192u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:06}").into_bytes(),
[b'v'; 100],
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
std::fs::File::open(&path)?.sync_all()?;
let table = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let punch = table.punch_offset_for(b"k004096")?;
let unit = super::inner::PUNCH_ALIGN;
assert!(
punch > 4 * unit,
"the fixture has a prefix of many allocation units"
);
let before = fs
.allocated_size(&path)?
.expect("a unix file reports st_blocks");
table.mark_punch_on_drop(punch);
drop(table);
let after = fs
.allocated_size(&path)?
.expect("a unix file reports st_blocks");
let freed = before
.checked_sub(after)
.expect("a punch allocates nothing");
assert!(
freed + unit >= punch,
"the reclaim freed {freed} of the {punch}-byte prefix ({before} -> {after} allocated)",
);
let reopened = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path, checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let geometry = reopened.punch_geometry()?;
assert!(
matches!(geometry.verdict, crate::table::PunchProbe::Punched),
"the filesystem's hole probe proves the punch",
);
match Table::conservative_restriction(&geometry) {
crate::table::DerivedRestriction::Bound(bound) => assert!(
bound.as_ref() >= b"k004096".as_slice(),
"a derived bound below the committed one would serve consumed keys",
),
_ => panic!("a clean punched prefix derives a bound"),
}
Ok(())
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_retains_the_reclaim_while_a_checkpoint_link_survives() -> crate::Result<()> {
use crate::fs::Fs;
let links = Arc::new(core::sync::atomic::AtomicU64::new(2));
let fs: Arc<dyn Fs> = Arc::new(SharedInodeFs(crate::fs::MemFs::new(), Arc::clone(&links)));
let root = std::path::absolute("/db")?;
fs.create_dir_all(&root)?;
let path = root.join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
for i in 0..256u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let pause = crate::deletion_pause::DeletionPause::new_shared();
table.install_deletion_pause(Arc::clone(&pause));
let punch = table.punch_offset_for(b"k0128")?;
assert!(punch > 0, "the fixture has a punchable prefix");
table.mark_punch_on_drop(punch);
drop(table);
assert!(
pause.has_pending_reclaims(),
"a reclaim blocked by a completed checkpoint's link must be retained \
for a retry, not discarded",
);
links.store(1, core::sync::atomic::Ordering::Release);
pause.retry_pending_reclaims();
assert!(
!pause.has_pending_reclaims(),
"an exclusively-owned file's retained reclaim is finished by the retry",
);
let after = {
let file = fs.open(&path, &crate::fs::FsOpenOptions::new().read(true))?;
let len = crate::fs::FsFile::metadata(&*file)?.len;
crate::file::read_exact(&*file, 0, usize::try_from(len).unwrap_or(0))?.to_vec()
};
assert!(
after
.get(..64)
.is_some_and(|head| head.iter().all(|&b| b == 0)),
"the retried reclaim punches the consumed prefix",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn punch_on_drop_waits_for_the_pin_listing_the_file() -> crate::Result<()> {
use crate::fs::Fs;
let fs: Arc<dyn Fs> = Arc::new(crate::fs::MemFs::new());
let root = std::path::absolute("/db")?;
fs.create_dir_all(&root)?;
let path = root.join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
for i in 0..256u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let read = |fs: &Arc<dyn Fs>| -> crate::Result<Vec<u8>> {
let file = fs.open(&path, &crate::fs::FsOpenOptions::new().read(true))?;
let len = crate::fs::FsFile::metadata(&*file)?.len;
Ok(crate::file::read_exact(&*file, 0, usize::try_from(len).unwrap_or(0))?.to_vec())
};
let before = read(&fs)?;
let pause = crate::deletion_pause::DeletionPause::new_shared();
table.install_deletion_pause(Arc::clone(&pause));
pause.pin_files(core::iter::once(path.as_path()));
let punch = table.punch_offset_for(b"k0128")?;
assert!(punch > 0, "the fixture has a punchable prefix");
table.mark_punch_on_drop(punch);
drop(table);
assert_eq!(read(&fs)?, before, "a pinned file keeps its bytes");
assert!(
pause.has_pending_reclaims(),
"the reclaim waits for the pin"
);
pause.unpin_files(core::iter::once(path.as_path()));
assert!(!pause.has_pending_reclaims(), "the pin's release ran it");
assert!(
read(&fs)?
.get(..64)
.is_some_and(|head| head.iter().all(|&b| b == 0)),
"the reclaim punched the consumed prefix once the pin dropped",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn deleting_a_pinned_table_keeps_its_heal_attestation() -> crate::Result<()> {
use crate::fs::Fs;
use std::io::Write;
let fs: Arc<dyn Fs> = Arc::new(crate::fs::MemFs::new());
let root = std::path::absolute("/db")?;
fs.create_dir_all(&root)?;
let path = root.join("0");
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?;
writer.write(crate::InternalValue::from_components(
b"k",
b"v",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer.finish()?.expect("table written");
let table = {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path.clone(), checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
let attestation = crate::scrub::heal_attest::attest_path(&path);
fs.open(
&attestation,
&crate::fs::FsOpenOptions::new().write(true).create(true),
)?
.write_all(b"attestation")?;
let pause = crate::deletion_pause::DeletionPause::new_shared();
table.install_deletion_pause(Arc::clone(&pause));
pause.pin_files(core::iter::once(path.as_path()));
table.mark_as_deleted();
drop(table);
assert!(fs.exists(&path)?, "the pinned table stays");
assert!(fs.exists(&attestation)?, "its attestation stays with it");
pause.unpin_files(core::iter::once(path.as_path()));
assert!(!fs.exists(&path)?, "the release removed the table");
assert!(!fs.exists(&attestation)?, "and its attestation");
Ok(())
}
#[cfg(all(feature = "std", feature = "page_ecc"))]
#[test]
fn reopen_restricted_propagates_the_shared_heal_and_deletion_gates() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let pause = crate::deletion_pause::DeletionPause::new_shared();
table.install_deletion_pause(std::sync::Arc::clone(&pause));
let lock = table.heal_lock_arc();
let restricted = table.reopen_restricted(crate::UserKey::from(&b"b"[..]))?;
let restricted_pause = restricted
.0
.deletion_pause
.get()
.expect("the deletion pause is propagated");
assert!(
std::sync::Arc::ptr_eq(restricted_pause, &pause),
"the restricted view shares the ORIGINAL deletion pause",
);
assert!(
std::sync::Arc::ptr_eq(&restricted.heal_lock_arc(), &lock),
"the restricted view shares the ORIGINAL heal lock",
);
Ok(())
},
None,
Some(|x| x),
)
}
#[cfg(feature = "page_ecc")]
#[test]
fn raw_block_parity_delta_rejects_a_trailer_length_mismatch() -> crate::Result<()> {
use crate::coding::Decode;
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let keyed = table
.data_block_handles()
.next()
.expect("a data block")
.expect("index entry decodes");
let handle = keyed.as_ref();
let file = std::fs::read(&*table.path)?;
let start = usize::try_from(handle.offset().0).unwrap_or(usize::MAX);
let Some(raw) = file.get(start..start + handle.size() as usize) else {
panic!("block frame within the file");
};
let header = crate::table::block::Header::decode_from(&mut &raw[..])?;
assert!(
matches!(table.raw_block_parity_delta(raw, &header), Ok(None)),
"the intact frame's parity trailer matches",
);
let Some(short) = raw.get(..raw.len() - 1) else {
panic!("frame is non-empty");
};
assert!(
table.raw_block_parity_delta(short, &header).is_err(),
"a trailer-length mismatch must be unverifiable, not healable",
);
Ok(())
},
None,
Some(|w: Writer| {
let Ok(params) = crate::table::block::EccParams::try_new(8, 2) else {
panic!("RS(8,2) params are valid");
};
w.use_ecc(Some(params))
}),
)
}
#[test]
fn scrub_block_rejects_a_block_of_the_wrong_role() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let outcome = crate::table::util::scrub_block(
table.global_id(),
&table.path,
&table.file_accessor,
&table.regions.tli,
crate::table::block::BlockType::Data,
table.metadata.data_block_compression,
table.encryption.as_deref(),
table.metadata.ecc_params,
#[cfg(zstd_any)]
table.zstd_dictionary.as_deref(),
None,
#[cfg(feature = "metrics")]
&table.metrics,
);
assert!(
matches!(outcome, Err(crate::Error::InvalidTag(("BlockType", _))),),
"a checksum-clean block of the WRONG role must fail the \
role check specifically (not scrub clean, and not fail for \
an unrelated reason): {outcome:?}",
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn salvage_load_block_reencodes_an_over_read_frame() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let keyed = table
.data_block_handles()
.next()
.expect("a data block")
.expect("index entry decodes");
let handle = keyed.as_ref();
let clean = table.salvage_load_block(handle, crate::table::block::BlockType::Data)?;
assert!(clean.verbatim.is_some(), "a clean read is verbatim-safe");
let over = crate::table::BlockHandle::new(handle.offset(), handle.size() + 8);
let sb = table.salvage_load_block(&over, crate::table::block::BlockType::Data)?;
assert!(
sb.verbatim.is_none(),
"an over-read frame with an opaque trailer must fall back to \
the re-encode path, not offer its overlong raw bytes for a \
verbatim copy",
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn reopen_restricted_carries_the_live_suffix_digest() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let restricted = table.reopen_restricted(crate::UserKey::from(&b"b"[..]))?;
let punch = table.punch_offset_for(b"b")?;
let expected = crate::Checksum::from_raw(crate::repair::compute_table_checksum_from(
&crate::fs::StdFs,
&table.path,
punch,
)?);
assert_eq!(
restricted.checksum(),
expected,
"the restricted view reports its live-suffix digest",
);
assert!(
punch > 0,
"reopening at a later block punches a real prefix"
);
assert_ne!(
restricted.checksum(),
table.checksum(),
"a non-zero punch offset makes the suffix digest differ from \
the whole-file digest",
);
Ok(())
},
Some(1),
Some(|x| x),
)
}
fn twelve_letter_items() -> Vec<crate::InternalValue> {
(b'a'..=b'l')
.map(|c| crate::InternalValue::from_components([c], b"v", 0, crate::ValueType::Value))
.collect()
}
fn reconcile_read_count(blocks: usize) -> crate::Result<(usize, usize)> {
use crate::fs::{FaultFs, StdFs};
let dir = tempfile::tempdir()?;
let faulty = std::sync::Arc::new(FaultFs::new(StdFs));
let injector = faulty.injector();
let path = dir.path().join("table");
let mut writer = Writer::new(path.clone(), 0, 0, faulty.clone())?.use_zone_map(true);
for idx in 0..blocks * 4 {
if idx % 4 == 0 {
writer.spill_block()?;
}
writer.write(crate::InternalValue::from_components(
alloc::format!("key{idx:04}").into_bytes(),
b"v".as_slice(),
0,
crate::ValueType::Value,
))?;
}
let Some((_, checksum)) = writer.finish()? else {
panic!("the fixture writes entries");
};
let mut params = test_recover_params(path, checksum);
params.fs = faulty;
let table = Table::recover(params)?;
assert_eq!(
table.block_index.iter().count(),
blocks,
"the fixture must produce one data block per four entries",
);
injector.clear();
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("a healthy table must pass every gate, {gate:?} refused it: {e}");
}
Ok((injector.read_count(), injector.open_count()))
}
#[test]
fn reconcile_gates_read_each_block_once() -> crate::Result<()> {
let (small, small_opens) = reconcile_read_count(3)?;
let (large, large_opens) = reconcile_read_count(6)?;
assert_eq!(
large - small,
3,
"three more blocks must cost three more reads, got {small} then {large}",
);
assert_eq!(
(small_opens, large_opens),
(1, 1),
"a reconcile must open the table once",
);
Ok(())
}
#[test]
fn live_item_count_drops_the_straddling_block_rows_below_the_bound() -> crate::Result<()> {
test_with_table(
&twelve_letter_items(),
|table| {
let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
assert!(
restricted.punch_offset_for(b"h")? > 0,
"the fixture must punch a real prefix, not the degenerate whole file",
);
assert_eq!(
5,
restricted.live_item_count(crate::table::util::ReadCharge::Maintenance)?,
"a zone-mapped view counts the straddling block's live suffix, \
not its whole row count",
);
Ok(())
},
Some(4),
Some(|w: Writer| w.use_zone_map(true)),
)
}
#[cfg(feature = "metrics")]
#[test]
fn live_item_count_for_maintenance_over_a_cold_straddling_block_counts_no_bytes()
-> crate::Result<()> {
test_with_table(
&twelve_letter_items(),
|table| {
let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
let metrics = &restricted.metrics;
let before = (
metrics.bytes_read(),
metrics.bytes_decoded(),
metrics.bytes_copied(),
);
assert_eq!(
5,
restricted.live_item_count(crate::table::util::ReadCharge::Maintenance)?
);
assert_eq!(
(
metrics.bytes_read(),
metrics.bytes_decoded(),
metrics.bytes_copied(),
),
before,
"a report's live-row count must not reach the read counters",
);
Ok(())
},
Some(4),
Some(|w: Writer| w.use_zone_map(true)),
)
}
#[cfg(feature = "metrics")]
#[test]
fn live_item_count_for_a_query_over_a_cold_straddling_block_counts_its_reads() -> crate::Result<()>
{
test_with_table(
&twelve_letter_items(),
|table| {
let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
let metrics = &restricted.metrics;
let (read, decoded) = (metrics.bytes_read(), metrics.bytes_decoded());
assert_eq!(
5,
restricted.live_item_count(crate::table::util::ReadCharge::Foreground)?
);
assert!(
metrics.bytes_read() > read && metrics.bytes_decoded() > decoded,
"the straddling block a query's estimate reads must be counted",
);
Ok(())
},
Some(4),
Some(|w: Writer| w.use_zone_map(true)),
)
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
fn ecc_two_level_table(
dir: &tempfile::TempDir,
cache: Arc<crate::Cache>,
) -> crate::Result<(Table, std::path::PathBuf)> {
let (file, checksum) = write_ecc_two_level_table(dir, false)?;
Ok((recover_with_cache(&file, checksum, cache)?, file))
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
fn recover_with_cache(
file: &std::path::Path,
checksum: crate::Checksum,
cache: Arc<crate::Cache>,
) -> crate::Result<Table> {
let mut params = test_recover_params(file.to_path_buf(), checksum);
params.cache = cache;
params.metrics = Arc::new(Metrics::default());
let table = Table::recover(params)?;
assert!(
table.metadata.index_block_count > 1,
"the fixture must split its index into partitions",
);
Ok(table)
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
fn write_ecc_two_level_table(
dir: &tempfile::TempDir,
locator: bool,
) -> crate::Result<(std::path::PathBuf, crate::Checksum)> {
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_ecc(Some(crate::table::block::EccParams::RS_4_2))
.use_partitioned_index()
.use_data_block_size(1)
.use_meta_partition_size(3);
if locator {
writer = writer.use_locator(crate::config::LocatorPolicyEntry::Enabled {
precision: crate::config::LocatorPrecision::Restart,
block_id_bits: None,
slot_bits: None,
});
}
for (i, key) in [b"a", b"b", b"c", b"d", b"e", b"f", b"g", b"h"]
.into_iter()
.enumerate()
{
writer.write(InternalValue::from_components(
key.as_slice(),
b"value-payload-bytes",
u64::try_from(i).expect("small") + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("the fixture writes entries");
Ok((file, checksum))
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
fn flip_payload_bit(table: &Table, file: &std::path::Path, offset: u64) -> crate::Result<()> {
let mut bytes = std::fs::read(file)?;
let pos = usize::try_from(offset).expect("offset fits usize")
+ crate::table::block::Header::MIN_LEN
+ 3;
bytes[pos] ^= 0x80;
std::fs::write(file, &bytes)?;
table.file_accessor.remove_for_table(&table.global_id());
Ok(())
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
#[test]
fn live_item_count_for_a_report_over_a_repaired_block_counts_the_repair() -> crate::Result<()> {
let dir = tempdir()?;
let (table, file) = ecc_two_level_table(&dir, Arc::new(crate::Cache::with_capacity_bytes(0)))?;
let restricted = table.with_restriction(crate::UserKey::from(&b"d"[..]));
let straddle = restricted.punch_offset_for(b"d")?;
assert!(straddle > 0, "the fixture must punch a real prefix");
flip_payload_bit(&table, &file, straddle)?;
let before = table.metrics.ecc_recovered_count();
assert_eq!(
5,
restricted.live_item_count(crate::table::util::ReadCharge::Untraced)?
);
assert!(
table.metrics.ecc_recovered_count() > before,
"the repair a report's read made must reach the ECC counters",
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
#[test]
fn a_maintenance_index_walk_over_a_repaired_partition_counts_the_repair() -> crate::Result<()> {
let dir = tempdir()?;
let (table, file) = ecc_two_level_table(&dir, Arc::new(crate::Cache::with_capacity_bytes(0)))?;
let last = table
.block_index
.iter()
.next_back()
.expect("the table has data blocks")?;
let first_partition = last.offset().0 + u64::from(last.size());
flip_payload_bit(&table, &file, first_partition)?;
let before = table.metrics.ecc_recovered_count();
table
.maintenance_index_walk()
.collect::<crate::Result<Vec<_>>>()?;
let after_maintenance = table.metrics.ecc_recovered_count();
table
.block_index
.iter()
.collect::<crate::Result<Vec<_>>>()?;
assert!(
table.metrics.ecc_recovered_count() > after_maintenance,
"the fixture must corrupt a partition a walk repairs",
);
assert!(
after_maintenance > before,
"the repair a maintenance walk made must reach the ECC counters",
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
#[test]
fn reconcile_gates_over_a_cold_index_count_no_bytes() -> crate::Result<()> {
let dir = tempdir()?;
let (file, checksum) = write_ecc_two_level_table(&dir, true)?;
let table = recover_with_cache(
&file,
checksum,
Arc::new(crate::Cache::with_capacity_bytes(0)),
)?;
assert!(
table.regions.locator.is_some(),
"the fixture must carry a locator for the gate to pair",
);
let metrics = &table.metrics;
let before = (metrics.bytes_read(), metrics.bytes_decoded());
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("a healthy table must pass every gate, {gate:?} refused it: {e}");
}
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
before,
"the reconcile gates' index walks must not reach the read counters",
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
#[test]
fn live_item_count_for_a_report_caches_nothing() -> crate::Result<()> {
let dir = tempdir()?;
let (file, checksum) = write_ecc_two_level_table(&dir, false)?;
let (straddle, first_partition) = {
let probe = recover_with_cache(
&file,
checksum,
Arc::new(crate::Cache::with_capacity_bytes(0)),
)?;
let last = probe
.block_index
.iter()
.next_back()
.expect("the table has data blocks")?;
(
probe.punch_offset_for(b"d")?,
last.offset().0 + u64::from(last.size()),
)
};
let table = recover_with_cache(
&file,
checksum,
Arc::new(crate::Cache::with_capacity_bytes(10_000_000)),
)?;
let restricted = table.with_restriction(crate::UserKey::from(&b"d"[..]));
assert_eq!(
5,
restricted.live_item_count(crate::table::util::ReadCharge::Untraced)?
);
let id = table.global_id();
assert!(
!table
.cache
.has_block(id, crate::table::BlockOffset(straddle)),
"a report must not cache the straddling block it read",
);
assert!(
!table
.cache
.has_block(id, crate::table::BlockOffset(first_partition)),
"a report must not cache the index partitions it walked",
);
Ok(())
}
#[test]
fn live_item_count_apportions_only_the_blocks_above_the_straddling_one() -> crate::Result<()> {
test_with_table(
&twelve_letter_items(),
|table| {
let restricted = table.with_restriction(crate::UserKey::from(&b"h"[..]));
let live = restricted.live_item_count(crate::table::util::ReadCharge::Maintenance)?;
assert!(
(4..=6).contains(&live),
"apportioning above the straddling block must land near the \
5 live entries, got {live}",
);
Ok(())
},
Some(4),
Some(|x| x),
)
}
#[test]
fn punch_offset_for_locates_the_first_block_reaching_a_key() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(0, table.punch_offset_for(b"a")?);
let pb = table.punch_offset_for(b"b")?;
let pc = table.punch_offset_for(b"c")?;
let pe = table.punch_offset_for(b"e")?;
assert!(pb > 0, "punching up to b reclaims a's block");
assert!(pc > pb, "offsets advance with the key");
assert!(pe > pc);
let beyond = table.punch_offset_for(b"zzz")?;
assert!(
beyond >= pe,
"a key beyond the last block punches every data block",
);
Ok(())
},
Some(1),
Some(|x| x),
)
}
#[cfg(test)]
#[expect(clippy::unwrap_used)]
fn recover_adaptive_table(
items: &[crate::InternalValue],
spill_threshold: u64,
) -> crate::Result<(Table, tempfile::TempDir)> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("table");
let mut writer =
Writer::new(path.clone(), 0, 0, Arc::new(StdFs))?.use_adaptive_index(spill_threshold);
for item in items {
writer.write(item.clone())?;
}
let (_, checksum) = writer.finish()?.unwrap();
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
#[cfg_attr(not(feature = "metrics"), expect(unused_mut))]
let mut params = test_recover_params(path, checksum);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)?
};
Ok((table, dir))
}
#[test]
#[expect(clippy::unwrap_used)]
fn adaptive_index_small_sst_is_single_level() -> crate::Result<()> {
let items: Vec<_> = (0u32..500)
.map(|i| {
let key = format!("key-{i:08}");
crate::InternalValue::from_components(
key.as_bytes(),
b"some-value-payload",
0,
crate::ValueType::Value,
)
})
.collect();
let (table, _dir) = recover_adaptive_table(&items, u64::MAX)?;
assert!(
table.regions.index.is_none(),
"small index must stay single-level (Full), got a two-level index region",
);
assert_eq!(
items.len(),
usize::try_from(table.metadata.item_count).unwrap()
);
for i in 0u32..500 {
let key = format!("key-{i:08}");
let got = table.get(key.as_bytes(), SeqNo::MAX, hash64(key.as_bytes()))?;
assert_eq!(
b"some-value-payload",
&*got.unwrap().value,
"single-level read mismatch for {key}",
);
}
Ok(())
}
#[test]
#[expect(clippy::unwrap_used)]
fn adaptive_index_zero_threshold_spills_to_two_level() -> crate::Result<()> {
let items: Vec<_> = (0u32..500)
.map(|i| {
let key = format!("key-{i:08}");
crate::InternalValue::from_components(
key.as_bytes(),
b"some-value-payload",
0,
crate::ValueType::Value,
)
})
.collect();
let (table, _dir) = recover_adaptive_table(&items, 0)?;
assert!(
table.regions.index.is_some(),
"zero threshold must spill to a two-level (partitioned) index",
);
assert_eq!(
items.len(),
usize::try_from(table.metadata.item_count).unwrap()
);
for i in 0u32..500 {
let key = format!("key-{i:08}");
let got = table.get(key.as_bytes(), SeqNo::MAX, hash64(key.as_bytes()))?;
assert_eq!(
b"some-value-payload",
&*got.unwrap().value,
"two-level read mismatch for {key}",
);
}
Ok(())
}
#[test]
fn table_point_read_index_block_restart_interval() -> crate::Result<()> {
let items: Vec<_> = (0u32..24)
.map(|i| {
let key = format!("adj:out:vertex-0001:edge-{i:04}");
let value = format!("value-{i:04}");
crate::InternalValue::from_components(
key.as_bytes(),
value.as_bytes(),
u64::from(i),
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
assert_eq!(
b"value-0011",
&*table
.get(
b"adj:out:vertex-0001:edge-0011",
SeqNo::MAX,
hash64(b"adj:out:vertex-0001:edge-0011"),
)?
.expect("test assertion: expected value for edge-0011")
.value,
);
let range = table
.range(
UserKey::from("adj:out:vertex-0001:edge-0008")
..=UserKey::from("adj:out:vertex-0001:edge-0012"),
)
.flatten()
.collect::<Vec<_>>();
assert_eq!(items[8..=12], range);
Ok(())
},
Some(1),
Some(|writer: Writer| {
writer
.use_data_block_size(128)
.use_index_block_restart_interval(4)
}),
)
}
#[test]
#[cfg(feature = "zstd")]
fn table_point_read_zstd_dictionary() -> crate::Result<()> {
let dict = Arc::new(make_test_dictionary());
let expected_dict_id = dict.id();
let compression = crate::CompressionType::zstd_dict(3, expected_dict_id)?;
let items = [
crate::InternalValue::from_components(
b"key-00001",
b"value-00001-padding-to-make-it-longer",
3,
crate::ValueType::Value,
),
crate::InternalValue::from_components(
b"key-00002",
b"value-00002-padding-to-make-it-longer",
2,
crate::ValueType::Value,
),
];
test_with_table_and_zstd_dictionary(
&items,
|table| {
assert!(matches!(
table.metadata.data_block_compression,
crate::CompressionType::ZstdDict { dict_id, .. } if dict_id == expected_dict_id
));
assert_eq!(items, &*table.iter().flatten().collect::<Vec<_>>());
assert_eq!(
b"value-00001-padding-to-make-it-longer",
&*table
.get(b"key-00001", SeqNo::MAX, hash64(b"key-00001"),)?
.expect("test assertion: expected value for key-00001")
.value,
);
Ok(())
},
None,
Some(|writer: Writer| writer.use_data_block_compression(compression)),
dict,
)
}
#[test]
fn table_range_exclusive_bounds() -> crate::Result<()> {
use core::ops::Bound::{Excluded, Included};
let items = [
crate::InternalValue::from_components(b"a", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"v", 0, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
let res = table
.range((Excluded(UserKey::from("b")), Included(UserKey::from("d"))))
.flatten()
.collect::<Vec<_>>();
assert_eq!(
items.iter().skip(2).take(2).cloned().collect::<Vec<_>>(),
&*res,
);
let res = table
.range((Excluded(UserKey::from("b")), Included(UserKey::from("d"))))
.rev()
.flatten()
.collect::<Vec<_>>();
assert_eq!(
items
.iter()
.skip(2)
.take(2)
.rev()
.cloned()
.collect::<Vec<_>>(),
&*res,
);
let res = table
.range((Excluded(UserKey::from("b")), Excluded(UserKey::from("d"))))
.flatten()
.collect::<Vec<_>>();
assert_eq!(
items.iter().skip(2).take(1).cloned().collect::<Vec<_>>(),
&*res,
);
let res = table
.range((Excluded(UserKey::from("b")), Excluded(UserKey::from("d"))))
.rev()
.flatten()
.collect::<Vec<_>>();
assert_eq!(
items
.iter()
.skip(2)
.take(1)
.rev()
.cloned()
.collect::<Vec<_>>(),
&*res,
);
Ok(())
},
None,
Some(|x: Writer| x.use_data_block_size(1)),
)
}
#[test]
fn writer_records_effective_page_ecc_descriptor() -> crate::Result<()> {
let items = [crate::InternalValue::from_components(
b"a",
b"v",
0,
crate::ValueType::Value,
)];
test_with_table(
&items,
|table| {
assert_eq!(
table.metadata.page_ecc,
cfg!(feature = "page_ecc"),
"descriptor#page_ecc must reflect the effective (compiled) page_ecc setting",
);
Ok(())
},
None,
Some(|w: Writer| {
w.use_page_ecc(
true,
crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 4,
parity_shards: 2,
},
)
}),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read_mvcc_block_boundary() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"5", 5, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"4", 4, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"3", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"2", 2, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"1", 1, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(2, table.metadata.data_block_count);
let key_hash = hash64(b"a");
assert_eq!(
b"5",
&*table.get(b"a", SeqNo::MAX, key_hash)?.unwrap().value
);
assert_eq!(b"4", &*table.get(b"a", 5, key_hash)?.unwrap().value);
assert_eq!(b"3", &*table.get(b"a", 4, key_hash)?.unwrap().value);
assert_eq!(b"2", &*table.get(b"a", 3, key_hash)?.unwrap().value);
assert_eq!(b"1", &*table.get(b"a", 2, key_hash)?.unwrap().value);
Ok(())
},
Some(3),
Some(|x| x),
)
}
#[test]
fn table_scan() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(items, &*table.scan()?.flatten().collect::<Vec<_>>());
assert_eq!(
table.metadata.key_range,
crate::KeyRange::new((b"abc".into(), b"xyz".into())),
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn table_iter_simple() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(items, &*table.iter().flatten().collect::<Vec<_>>());
assert_eq!(
items.iter().rev().cloned().collect::<Vec<_>>(),
&*table.iter().rev().flatten().collect::<Vec<_>>(),
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn table_range_simple() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"abc", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"def", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"xyz", b"asdasdasd", 3, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(
items.iter().skip(1).cloned().collect::<Vec<_>>(),
&*table
.range(UserKey::from("b")..)
.flatten()
.collect::<Vec<_>>()
);
assert_eq!(
items.iter().skip(1).rev().cloned().collect::<Vec<_>>(),
&*table
.range(UserKey::from("b")..)
.rev()
.flatten()
.collect::<Vec<_>>(),
);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn table_range_ping_pong() -> crate::Result<()> {
let items = (0u64..10)
.map(|i| InternalValue::from_components(i.to_be_bytes(), "", 0, crate::ValueType::Value))
.collect::<Vec<_>>();
test_with_table(
&items,
|table| {
let mut iter =
table.range(UserKey::from(5u64.to_be_bytes())..UserKey::from(10u64.to_be_bytes()));
let mut count = 0;
for x in 0.. {
if x % 2 == 0 {
let Some(_) = iter.next() else {
break;
};
count += 1;
} else {
let Some(_) = iter.next_back() else {
break;
};
count += 1;
}
}
assert_eq!(5, count);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn table_range_multiple_data_blocks() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"asdasdasd", 3, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(5, table.metadata.data_block_count);
assert_eq!(
items.iter().skip(1).take(3).cloned().collect::<Vec<_>>(),
&*table
.range(UserKey::from("b")..=UserKey::from("d"))
.flatten()
.collect::<Vec<_>>()
);
assert_eq!(
items
.iter()
.skip(1)
.take(3)
.rev()
.cloned()
.collect::<Vec<_>>(),
&*table
.range(UserKey::from("b")..=UserKey::from("d"))
.rev()
.flatten()
.collect::<Vec<_>>(),
);
Ok(())
},
None,
Some(|x: Writer| x.use_data_block_size(1)),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_point_read_partitioned_filter_smoke_test() -> crate::Result<()> {
let items = [
crate::InternalValue::from_components(b"a", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"b", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"c", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"d", b"asdasdasd", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"e", b"asdasdasd", 3, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(1, table.metadata.data_block_count);
for item in &items {
let key_hash = hash64(&item.key.user_key);
assert_eq!(
item.value,
table
.get(&item.key.user_key, SeqNo::MAX, key_hash)
.unwrap()
.unwrap()
.value,
);
}
Ok(())
},
None,
Some(|x: Writer| x.use_partitioned_filter()),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_partitioned_filter() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a", "a7", 7, Value),
InternalValue::from_components("a", "a6", 6, Value),
InternalValue::from_components("a", "a5", 5, Value),
InternalValue::from_components("a", "a4", 4, Value),
InternalValue::from_components("a", "a3", 3, Value),
InternalValue::from_components("b", "b5", 5, Value),
InternalValue::from_components("c", "c8", 8, Value),
InternalValue::from_components("d", "d10", 10, Value),
];
test_with_table(
&items,
|table| {
assert!(table.regions.filter.is_some(), "filter should exist");
assert!(
table.regions.filter_tli.is_some(),
"filter TLI should exist"
);
assert_eq!(b"a7", &*table.get(b"a", 8, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a6", &*table.get(b"a", 7, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a5", &*table.get(b"a", 6, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a4", &*table.get(b"a", 5, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a3", &*table.get(b"a", 4, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"b5", &*table.get(b"b", 6, hash64(b"b"))?.unwrap().value,);
assert_eq!(b"c8", &*table.get(b"c", 9, hash64(b"c"))?.unwrap().value,);
assert_eq!(b"d10", &*table.get(b"d", 11, hash64(b"d"))?.unwrap().value,);
Ok(())
},
None,
Some(|x: Writer| x.use_partitioned_filter().use_meta_partition_size(3)),
)
}
#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_propagates_a_faulted_bloom_probe() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
use crate::io::ErrorKind;
let items: Vec<InternalValue> = (0..64u32)
.map(|i| {
InternalValue::from_components(
format!("key{i:04}").into_bytes(),
b"v".to_vec(),
1,
crate::ValueType::Value,
)
})
.collect();
let dir = tempdir()?;
let file = dir.path().join("table");
let injector = Arc::new(FaultInjector::new());
let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));
let checksum = {
let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?
.use_partitioned_filter()
.use_meta_partition_size(8);
for item in &items {
writer.write(item.clone())?;
}
writer.finish()?.unwrap().1
};
let table = {
let mut params = test_recover_params(file, checksum);
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));
let key = b"key0000".as_slice();
let sorted = [(key, hash64(key))];
let mut tally = crate::table::probe_stats::PlanCounts::default();
let result = table.plan_block_tasks(&sorted, SeqNo::MAX, &mut tally);
assert!(
result.is_err(),
"a faulted bloom-probe read must surface as Err, not a swallowed miss"
);
Ok(())
}
#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_propagates_a_faulted_index_read() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
use crate::io::ErrorKind;
let items: Vec<InternalValue> = (0u32..500)
.map(|i| {
InternalValue::from_components(
format!("key{i:06}").into_bytes(),
b"v".to_vec(),
1,
crate::ValueType::Value,
)
})
.collect();
let dir = tempdir()?;
let file = dir.path().join("table");
let injector = Arc::new(FaultInjector::new());
let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));
let checksum = {
let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?.use_adaptive_index(0);
for item in &items {
writer.write(item.clone())?;
}
writer.finish()?.unwrap().1
};
let table = {
let mut params = test_recover_params(file, checksum);
params.fs = Arc::clone(&fs);
params.pin_filter = true;
Table::recover(params)?
};
injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));
let key = b"key000250".as_slice();
let sorted = [(key, hash64(key))];
let mut tally = crate::table::probe_stats::PlanCounts::default();
let result = table.plan_block_tasks(&sorted, SeqNo::MAX, &mut tally);
assert!(
result.is_err(),
"a faulted block-index read must surface as Err, not a swallowed end-of-index"
);
Ok(())
}
#[test]
#[expect(clippy::unwrap_used, reason = "test code")]
fn plan_block_tasks_returns_none_for_a_table_above_the_snapshot() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultInjector, FaultOp, FaultRule};
use crate::io::ErrorKind;
let items: Vec<InternalValue> = (0u32..16)
.map(|i| {
InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v".to_vec(),
10,
crate::ValueType::Value,
)
})
.collect();
let dir = tempdir()?;
let file = dir.path().join("table");
let injector = Arc::new(FaultInjector::new());
let fs: Arc<dyn crate::fs::Fs> = Arc::new(FaultFs::with_injector(StdFs, Arc::clone(&injector)));
let checksum = {
let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?;
for item in &items {
writer.write(item.clone())?;
}
writer.finish()?.unwrap().1
};
let table = {
let mut params = test_recover_params(file, checksum);
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).on_path("table"));
let key = b"k0000".as_slice();
let sorted = [(key, hash64(key))];
let mut tally = crate::table::probe_stats::PlanCounts::default();
assert!(table.plan_block_tasks(&sorted, 5, &mut tally)?.is_none());
Ok(())
}
#[test]
fn table_seqnos() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a", nanoid::nanoid!().as_bytes(), 7, Value),
InternalValue::from_components("b", nanoid::nanoid!().as_bytes(), 5, Value),
InternalValue::from_components("c", nanoid::nanoid!().as_bytes(), 8, Value),
InternalValue::from_components("d", nanoid::nanoid!().as_bytes(), 10, Value),
];
test_with_table(
&items,
|table| {
assert_eq!(5, table.metadata.seqnos.0);
assert_eq!(10, table.metadata.seqnos.1);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn table_zero_bpk() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a", nanoid::nanoid!().as_bytes(), 7, Value),
InternalValue::from_components("b", nanoid::nanoid!().as_bytes(), 5, Value),
InternalValue::from_components("c", nanoid::nanoid!().as_bytes(), 8, Value),
InternalValue::from_components("d", nanoid::nanoid!().as_bytes(), 10, Value),
];
test_with_table(
&items,
|table| {
assert!(table.regions.filter.is_none());
Ok(())
},
None,
Some(|x: Writer| x.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(0.0))),
)
}
#[test]
#[expect(
clippy::unreadable_literal,
clippy::unwrap_used,
clippy::indexing_slicing,
clippy::cast_possible_truncation
)]
#[cfg(not(feature = "metrics"))]
fn table_read_fuzz_1() -> crate::Result<()> {
use crate::Slice;
use crate::ValueType::{Tombstone, Value};
let items = [
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
18340908174618760209,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
18054235897395861447,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([103]),
17820711698989577060,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
17652351990810576660,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
17576667967203573449,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([30]),
16889403751796995588,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([186]),
15595956295177086731,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
15512796775024989213,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([188, 156, 59, 85, 13]),
15149465603839159843,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([174, 71]),
15102256701513339307,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([35, 148]),
15091160407760527013,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
14675333203365509622,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([245]),
14571905818510788533,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
14541113699969547298,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
14486387191240337417,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
14112006182482717758,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([159]),
13992512869528291746,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
13915106262991388976,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
13597506620670366065,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
13064400463180401957,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
12969967266897711474,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
12508372658468564628,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([138]),
11795269606598686255,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([18]),
10730214428751858128,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([236]),
10124645034840293700,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([216, 81]),
9559308046784608794,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([79]),
8607115510826103394,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
7963767336149785641,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
7882646634183551394,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
7719307175583565930,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([111]),
7522791039398476411,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([227, 164, 129]),
7410771579448817672,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
7003757491682295965,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
5723101273557106371,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
5581364419922287132,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([119, 29]),
5541782075650463683,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
5136199042703471864,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
5051972816573966850,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([162]),
5020119417385108821,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([69]),
4325966282181409009,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
4238714774310338082,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
4200824275757201410,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([92, 145, 251, 240, 133]),
3894954012280195585,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([14]),
3814525464013269105,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
3766663710061910506,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([129]),
3749655073597306832,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([231]),
3319226033273656005,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
3274394613296787928,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
2045761581956846404,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([78]),
1704041985603476880,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([]),
1441130125005023946,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([164, 136]),
1225420702887300153,
Tombstone,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([55]),
974698856173325051,
Value,
),
InternalValue::from_components(
Slice::from([0]),
Slice::from([238, 237]),
47340610649818236,
Value,
),
InternalValue::from_components(Slice::from([0]), Slice::from([]), 0, Value),
InternalValue::from_components(
Slice::from([0, 161]),
Slice::from([]),
17872519117933825384,
Tombstone,
),
InternalValue::from_components(
Slice::from([0, 161]),
Slice::from([]),
4494664966150999400,
Tombstone,
),
InternalValue::from_components(
Slice::from([1]),
Slice::from([]),
15373275907316083975,
Value,
),
];
let dir = tempfile::tempdir()?;
let file = dir.path().join("table_fuzz");
let data_block_size = 97;
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
.unwrap()
.use_data_block_size(data_block_size);
for item in items.iter().cloned() {
writer.write(item).unwrap();
}
let _trailer = writer.finish().unwrap();
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
params.pin_filter = true;
params.pin_index = true;
crate::Table::recover(params).unwrap()
};
let item_count_usize = table.metadata.item_count as usize;
assert_eq!(item_count_usize, items.len());
assert_eq!(items.len(), item_count_usize);
let items = items.into_iter().collect::<Vec<_>>();
assert_eq!(items, table.iter().collect::<Result<Vec<_>, _>>().unwrap());
assert_eq!(
items.iter().rev().cloned().collect::<Vec<_>>(),
table.iter().rev().collect::<Result<Vec<_>, _>>().unwrap(),
);
{
let lo = 0;
let hi = 54;
let lo_key = &items[lo].key.user_key;
let hi_key = &items[hi].key.user_key;
assert_eq!(lo_key, hi_key);
let expected_range: Vec<_> = items[lo..=hi].to_vec();
let iter = table.range(lo_key..=hi_key);
assert_eq!(expected_range, iter.collect::<Result<Vec<_>, _>>().unwrap());
}
Ok(())
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_partitioned_index() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a", "a7", 7, Value),
InternalValue::from_components("a", "a6", 6, Value),
InternalValue::from_components("a", "a5", 5, Value),
InternalValue::from_components("a", "a4", 4, Value),
InternalValue::from_components("a", "a3", 3, Value),
InternalValue::from_components("b", "b5", 5, Value),
InternalValue::from_components("c", "c8", 8, Value),
InternalValue::from_components("d", "d10", 10, Value),
];
let dir = tempfile::tempdir()?;
let file = dir.path().join("table_fuzz");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
.unwrap()
.use_partitioned_index()
.use_data_block_size(5)
.use_meta_partition_size(3);
for item in items.iter().cloned() {
writer.write(item).unwrap();
}
let _trailer = writer.finish().unwrap();
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
params.pin_filter = true;
params.pin_index = true;
crate::Table::recover(params).unwrap()
};
assert!(
table.regions.index.is_some(),
"2nd-level index should exist",
);
assert!(
table.metadata.index_block_count > 1,
"should use partitioned index",
);
assert_eq!(b"a7", &*table.get(b"a", 8, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a6", &*table.get(b"a", 7, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a5", &*table.get(b"a", 6, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a4", &*table.get(b"a", 5, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"a3", &*table.get(b"a", 4, hash64(b"a"))?.unwrap().value,);
assert_eq!(b"b5", &*table.get(b"b", 6, hash64(b"b"))?.unwrap().value,);
assert_eq!(b"c8", &*table.get(b"c", 9, hash64(b"c"))?.unwrap().value,);
assert_eq!(b"d10", &*table.get(b"d", 11, hash64(b"d"))?.unwrap().value,);
Ok(())
}
#[test]
fn seqno_range_reports_none_when_the_base_pushes_it_past_the_seqno_space() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a", "a", 0, Value),
InternalValue::from_components("b", "b", 8, Value),
];
let dir = tempfile::tempdir()?;
let file = dir.path().join("seqno-range");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for item in items.iter().cloned() {
writer.write(item)?;
}
let _trailer = writer.finish()?;
let recover = |base: crate::SeqNo| -> crate::Result<crate::Table> {
let mut params = test_recover_params(file.clone(), crate::Checksum::from_raw(0));
params.global_seqno = base;
crate::Table::recover(params)
};
assert_eq!(
Some((7, 15)),
recover(7)?.seqno_range(),
"a base that fits shifts both bounds by it",
);
assert_eq!(
None,
recover(crate::SeqNo::MAX - 4)?.seqno_range(),
"a base that does not fit reports no range rather than a wrapped one",
);
Ok(())
}
#[test]
#[expect(clippy::unwrap_used)]
fn table_global_seqno() -> crate::Result<()> {
use crate::ValueType::Value;
let items = [
InternalValue::from_components("a0", "a0", 0, Value),
InternalValue::from_components("a1", "a1", 1, Value),
InternalValue::from_components("b", "b", 8, Value),
];
let dir = tempfile::tempdir()?;
let file = dir.path().join("table_fuzz");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))
.unwrap()
.use_partitioned_filter()
.use_data_block_size(1)
.use_meta_partition_size(1);
for item in items.iter().cloned() {
writer.write(item).unwrap();
}
let _trailer = writer.finish().unwrap();
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.global_seqno = 7;
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
params.pin_filter = true;
params.pin_index = true;
crate::Table::recover(params).unwrap()
};
assert!(table.get(b"a1", 8, hash64(b"a1"))?.is_none());
assert_eq!(b"a0", &*table.get(b"a0", 8, hash64(b"a0"))?.unwrap().value,);
Ok(())
}
#[test]
#[expect(clippy::unwrap_used, reason = "test assertions")]
fn table_return_global_seqno() -> crate::Result<()> {
use crate::ValueType::Value;
use crate::fs::StdFs;
const SEQNO: SeqNo = 15;
let items = [InternalValue::from_components("abc", "abc", 0, Value)];
let dir = tempfile::tempdir()?;
let file = dir.path().join("table_fuzz");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for item in items {
writer.write(item)?;
}
let _trailer = writer.finish()?;
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.global_seqno = SEQNO;
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
params.pin_filter = true;
params.pin_index = true;
crate::Table::recover(params)?
};
assert_eq!(
InternalValue::from_components("abc", "abc", SEQNO, Value),
table.get(b"abc", 2 * SEQNO, hash64(b"abc"))?.unwrap(),
);
Ok(())
}
#[test]
fn scan_seqno_range_returns_nothing_below_a_bulk_ingest_base() -> crate::Result<()> {
use crate::ValueType::Value;
use crate::fs::StdFs;
const BASE: SeqNo = 100;
let dir = tempfile::tempdir()?;
let file = dir.path().join("ingested");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components("abc", "abc", 0, Value))?;
let _trailer = writer.finish()?;
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.global_seqno = BASE;
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
crate::Table::recover(params)?
};
assert_eq!(
table.scan_seqno_range(0, BASE / 2, true, (Bound::Unbounded, Bound::Unbounded))?,
Vec::new(),
"an upper bound below the ingest base must exclude the whole table",
);
assert_eq!(
table.scan_seqno_range(0, BASE, true, (Bound::Unbounded, Bound::Unbounded))?,
vec![InternalValue::from_components("abc", "abc", BASE, Value)],
"a window covering the base returns the row at its effective seqno",
);
Ok(())
}
#[test]
fn key_versions_leaves_out_a_version_whose_translated_seqno_overflows() -> crate::Result<()> {
use crate::ValueType::Value;
use crate::fs::StdFs;
const BASE: SeqNo = SeqNo::MAX - 2;
let dir = tempfile::tempdir()?;
let file = dir.path().join("ingested");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components("abc", "newer", 5, Value))?;
writer.write(InternalValue::from_components("abc", "older", 0, Value))?;
let _trailer = writer.finish()?;
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.global_seqno = BASE;
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
crate::Table::recover(params)?
};
let (_, handles) = table.key_blocks(b"abc")?;
let mut versions = Vec::new();
let mut found = false;
for handle in &handles {
let block = table
.load_data_block(handle, &crate::table::util::LookupFd::default())?
.expect("a row table's block");
let (taken, in_block) = table.key_versions(&block, b"abc", SeqNo::MAX);
versions.extend(taken);
found |= in_block;
}
assert!(found, "the table holds the key");
assert_eq!(
versions,
vec![InternalValue::from_components("abc", "older", BASE, Value)],
"only the version a snapshot can see, at its translated seqno"
);
Ok(())
}
#[test]
fn scan_seqno_range_of_a_restricted_view_starts_at_the_bound() -> crate::Result<()> {
let key = |i: u32| format!("key{i:03}").into_bytes();
let items: Vec<_> = (0..40u32)
.map(|i| {
InternalValue::from_components(key(i), b"v".as_slice(), 1, crate::ValueType::Value)
})
.collect();
test_with_table(
&items,
|table| {
let restricted = table.reopen_restricted(crate::UserKey::from(key(20)))?;
let keys = |bounds: (Bound<&[u8]>, Bound<&[u8]>)| -> crate::Result<Vec<Vec<u8>>> {
Ok(restricted
.scan_seqno_range(0, SeqNo::MAX, true, bounds)?
.into_iter()
.map(|entry| entry.key.user_key.to_vec())
.collect())
};
let whole = keys((Bound::Unbounded, Bound::Unbounded))?;
assert_eq!(whole, (20..40).map(key).collect::<Vec<_>>());
let lo = key(5);
let hi = key(25);
let bounded = keys((Bound::Included(&lo), Bound::Included(&hi)))?;
assert_eq!(bounded.first(), Some(&key(20)), "{bounded:?}");
assert!(bounded.contains(&key(25)), "{bounded:?}");
Ok(())
},
None,
Some(|w: Writer| w.use_data_block_size(64)),
)
}
#[cfg(feature = "columnar")]
#[test]
fn scan_seqno_range_skips_wholly_deleted_columnar_groups() -> crate::Result<()> {
use crate::config::DeleteStrategy;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(128)
.delete_strategy(DeleteStrategy::MergeOnRead);
const ROWS: u32 = 64;
for i in 0..ROWS {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
for row in 0..ROWS - 1 {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let table = Table::recover(test_recover_params(file, checksum))?;
assert!(table.metadata.data_block_count > 1, "several groups");
let rows = table.scan_seqno_range(0, SeqNo::MAX, true, (Bound::Unbounded, Bound::Unbounded))?;
let keys: Vec<_> = rows.iter().map(|row| row.key.user_key.to_vec()).collect();
assert_eq!(keys, vec![format!("key{:05}", ROWS - 1).into_bytes()]);
Ok(())
}
#[expect(
clippy::expect_used,
reason = "test helper: data length is controlled and fits in u32"
)]
fn rt_block(data: Vec<u8>) -> Block {
let data_length = u32::try_from(data.len()).expect("test buffer fits in u32");
Block {
header: block::Header {
data_length,
uncompressed_length: data_length,
..block::Header::test_dummy(block::BlockType::RangeTombstone)
},
data: data.into(),
}
}
fn assert_rt_decode_error(data: Vec<u8>, expected_field: &str, expected_offset: u64) {
let block = rt_block(data);
match Table::decode_range_tombstones(&block, &crate::comparator::DefaultUserComparator) {
Err(crate::Error::RangeTombstoneDecode { field, offset }) => {
assert_eq!(
field, expected_field,
"expected field '{expected_field}', got '{field}'"
);
assert_eq!(
offset, expected_offset,
"expected offset {expected_offset}, got {offset}"
);
}
other => panic!(
"expected RangeTombstoneDecode {{ field: \"{expected_field}\" }}, got: {other:?}"
),
}
}
#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_invalid_interval_returns_error() {
use crate::io::{LE, WriteBytesExt};
let mut buf = Vec::new();
buf.write_u16::<LE>(1).unwrap(); buf.extend_from_slice(b"z");
buf.write_u16::<LE>(1).unwrap(); buf.extend_from_slice(b"a");
buf.write_u64::<LE>(1).unwrap();
assert_rt_decode_error(buf, "interval", 0);
}
#[test]
fn decode_range_tombstones_truncated_start_len_returns_error() {
assert_rt_decode_error(vec![0x01], "start_len", 0);
}
#[test]
fn decode_range_tombstones_empty_block_returns_error() {
assert_rt_decode_error(Vec::new(), "start_len", 0);
}
#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_start_len_exceeds_remaining_returns_error() {
use crate::io::{LE, WriteBytesExt};
let mut buf = Vec::new();
buf.write_u16::<LE>(100).unwrap();
buf.push(0xFF);
assert_rt_decode_error(buf, "start_len", 0);
}
#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_truncated_end_len_returns_error() {
use crate::io::{LE, WriteBytesExt};
let mut buf = Vec::new();
buf.write_u16::<LE>(1).unwrap(); buf.push(b'a'); buf.push(0x01);
assert_rt_decode_error(buf, "end_len", 3);
}
#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_end_len_exceeds_remaining_returns_error() {
use crate::io::{LE, WriteBytesExt};
let mut buf = Vec::new();
buf.write_u16::<LE>(1).unwrap(); buf.push(b'a'); buf.write_u16::<LE>(100).unwrap(); buf.push(0xFF);
assert_rt_decode_error(buf, "end_len", 3);
}
#[test]
#[expect(clippy::unwrap_used)]
fn decode_range_tombstones_truncated_seqno_returns_error() {
use crate::io::{LE, WriteBytesExt};
let mut buf = Vec::new();
buf.write_u16::<LE>(1).unwrap(); buf.push(b'a'); buf.write_u16::<LE>(1).unwrap(); buf.push(b'z'); buf.extend_from_slice(&[0x01, 0x00, 0x00, 0x00]);
assert_rt_decode_error(buf, "seqno", 6);
}
#[test]
#[cfg(feature = "metrics")]
fn load_block_range_tombstone_metrics() -> crate::Result<()> {
use crate::{
CompressionType,
cache::Cache,
range_tombstone::RangeTombstone,
table::{block::BlockType, util::load_block},
};
use core::sync::atomic::Ordering::Relaxed;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components(
b"a",
b"v1",
1,
crate::ValueType::Value,
))?;
writer.write(InternalValue::from_components(
b"z",
b"v2",
2,
crate::ValueType::Value,
))?;
writer.write_range_tombstone(RangeTombstone::new(b"b".into(), b"y".into(), 3));
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing data items"
)]
let (_, checksum) = writer.finish()?.unwrap();
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file, checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
#[cfg(feature = "metrics")]
{
params.metrics = metrics.clone();
}
Table::recover(params)?
};
let rt_handle = table
.regions
.range_tombstones
.expect("table should have range tombstone block");
let table_id = table.global_id();
assert_eq!(0, metrics.range_tombstone_block_load_io.load(Relaxed));
let fresh_cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
let _block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&rt_handle,
BlockType::RangeTombstone,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
assert_eq!(1, metrics.range_tombstone_block_load_io.load(Relaxed));
assert_eq!(0, metrics.range_tombstone_block_load_cached.load(Relaxed));
assert!(metrics.range_tombstone_block_io_requested.load(Relaxed) > 0);
assert_eq!(0, metrics.data_block_load_io.load(Relaxed));
let _block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&rt_handle,
BlockType::RangeTombstone,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
assert_eq!(1, metrics.range_tombstone_block_load_io.load(Relaxed));
assert_eq!(1, metrics.range_tombstone_block_load_cached.load(Relaxed));
assert_eq!(0, metrics.data_block_load_cached.load(Relaxed));
Ok(())
}
#[test]
fn load_block_cache_hit_rejects_wrong_block_type() -> crate::Result<()> {
use crate::{
CompressionType,
cache::Cache,
table::{block::BlockType, util::load_block},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components(
b"a",
b"v1",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
#[cfg(feature = "metrics")]
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file, checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
#[cfg(feature = "metrics")]
{
params.metrics = metrics.clone();
}
Table::recover(params)?
};
let table_id = table.global_id();
let tli_handle = table.regions.tli;
let fresh_cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
let _block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&tli_handle,
BlockType::Index,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
let result = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&tli_handle,
BlockType::Data,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
);
assert!(
matches!(&result, Err(crate::Error::InvalidTag(("BlockType", _)))),
"expected InvalidTag for block type mismatch on cache hit, got Ok or wrong Err",
);
Ok(())
}
#[test]
fn a_paced_block_read_paces_each_portion_it_reads_from_the_file() -> crate::Result<()> {
use crate::{
CompressionType,
cache::Cache,
table::{block::BlockType, util::load_block_paced},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components(
b"a",
b"v1",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let file_len = std::fs::metadata(&file)?.len();
let table = Table::recover(test_recover_params(file, checksum))?;
struct Recorder(std::sync::Mutex<Vec<(u64, u64)>>);
impl crate::table::util::ReadPacer for Recorder {
fn active(&self) -> bool {
true
}
fn pace(&self, offset: u64, len: u64) {
self.0
.lock()
.expect("lock is not poisoned")
.push((offset, len));
}
}
let pace = Recorder(std::sync::Mutex::new(Vec::new()));
let taken = || core::mem::take(&mut *pace.0.lock().expect("lock is not poisoned"));
let cache = Cache::with_capacity_bytes(10_000_000);
let read = |handle: &BlockHandle| {
load_block_paced(
table.global_id(),
&table.path,
&table.file_accessor,
&cache,
handle,
BlockType::Index,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
#[cfg(feature = "metrics")]
&table.metrics,
crate::table::util::ReadCharge::Foreground,
Some(&pace),
&crate::table::util::LookupFd::default(),
)
};
let tli = table.regions.tli;
read(&tli)?;
assert_eq!(taken(), vec![(*tli.offset(), u64::from(tli.size()))]);
read(&tli)?;
assert_eq!(
taken(),
vec![],
"a cached block is not read again, so not paced again"
);
let past_end = BlockHandle::new(crate::table::BlockOffset(file_len - 100), 4096);
assert!(read(&past_end).is_err(), "the read runs out of file");
assert_eq!(
taken(),
vec![(file_len - 100, 4096)],
"the portion is paced from where it starts; the pacer cuts it at the end"
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn a_block_rejected_for_its_role_counts_what_its_transform_decoded() -> crate::Result<()> {
use crate::{
CompressionType,
cache::Cache,
table::{block::BlockType, util::load_block},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components(
b"a",
b"v1",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file, checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
params.metrics = metrics.clone();
Table::recover(params)?
};
let decoded_before = metrics.bytes_decoded();
let result = load_block(
table.global_id(),
&table.path,
&table.file_accessor,
&Cache::with_capacity_bytes(10_000_000),
&table.regions.tli,
BlockType::Data,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
&metrics,
crate::table::util::ReadCharge::Foreground,
);
assert!(
matches!(&result, Err(crate::Error::InvalidTag(("BlockType", _)))),
"the index block must be refused as a data block",
);
assert!(
metrics.bytes_decoded() > decoded_before,
"the transform ran before the role check, so its output must be counted",
);
Ok(())
}
#[cfg(feature = "metrics")]
fn one_row_table_and_its_index_frame(
dir: &tempfile::TempDir,
) -> crate::Result<(Table, Arc<crate::metrics::Metrics>, Vec<u8>)> {
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(InternalValue::from_components(
b"a",
b"v1",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.cache = Arc::new(crate::cache::Cache::with_capacity_bytes(10_000_000));
params.metrics = metrics.clone();
Table::recover(params)?
};
let tli = table.regions.tli;
let start = usize::try_from(tli.offset().0).expect("block offset fits usize");
let end = start + tli.size() as usize;
let frame = std::fs::read(&file)?
.get(start..end)
.expect("the index block lies within the file")
.to_vec();
Ok((table, metrics, frame))
}
#[cfg(feature = "metrics")]
fn index_frame_decoded_len(table: &Table) -> crate::Result<u64> {
let block = crate::table::util::load_block(
table.global_id(),
&table.path,
&table.file_accessor,
&crate::cache::Cache::with_capacity_bytes(10_000_000),
&table.regions.tli,
crate::table::block::BlockType::Index,
crate::CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
&crate::metrics::Metrics::default(),
crate::table::util::ReadCharge::Foreground,
)?;
Ok(block.data.len() as u64)
}
#[cfg(feature = "metrics")]
#[test]
fn a_block_refused_for_its_declared_length_counts_what_its_transform_decoded() -> crate::Result<()>
{
use crate::{
CompressionType,
cache::Cache,
coding::{Decode, Encode},
table::{
block::{BlockType, Header},
util::load_block,
},
};
use std::io::{Seek, Write};
let dir = tempdir()?;
let (table, metrics, frame) = one_row_table_and_its_index_frame(&dir)?;
let produced = index_frame_decoded_len(&table)?;
let header_len = Header::header_len(BlockType::Index);
let mut header = Header::decode_from(&mut &frame[..header_len])?;
header.uncompressed_length += 1;
let mut tampered = header.encode_into_vec();
tampered.extend_from_slice(&frame[header_len..]);
assert_eq!(tampered.len(), frame.len(), "the frame keeps its size");
let mut wf = std::fs::OpenOptions::new().write(true).open(&*table.path)?;
wf.seek(std::io::SeekFrom::Start(table.regions.tli.offset().0))?;
wf.write_all(&tampered)?;
wf.sync_all()?;
let decoded_before = metrics.bytes_decoded();
let result = load_block(
table.global_id(),
&table.path,
&table.file_accessor,
&Cache::with_capacity_bytes(10_000_000),
&table.regions.tli,
BlockType::Index,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
&metrics,
crate::table::util::ReadCharge::Foreground,
);
assert!(result.is_err(), "the length mismatch refuses the block");
assert_eq!(
metrics.bytes_decoded() - decoded_before,
produced,
"the file read's transform ran, so its output must be counted",
);
let decoded_before = metrics.bytes_decoded();
assert!(
table
.decode_block_from_bytes(&tampered, table.regions.tli.offset().0, BlockType::Index)
.is_err(),
"the staged read refuses the block too",
);
assert_eq!(
metrics.bytes_decoded() - decoded_before,
produced,
"the staged read's transform ran, so its output must be counted",
);
let decoded_before = metrics.bytes_decoded();
assert!(
table
.decode_data_block_keeping(&tampered, &table.regions.tli, &mut 0, None)
.is_err(),
"the chunked resolver refuses the block too",
);
assert_eq!(
metrics.bytes_decoded() - decoded_before,
produced,
"the chunked resolver's transform ran, so its output must be counted",
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn a_confirming_reread_that_never_issues_counts_no_bytes() -> crate::Result<()> {
use crate::{
CompressionType,
heal_hints::HealHints,
table::{
block::BlockType,
util::{ReadCharge, maybe_record_persistent_heal},
},
};
let dir = tempdir()?;
let (table, metrics, _frame) = one_row_table_and_its_index_frame(&dir)?;
table.file_accessor.remove_for_table(&table.global_id());
let missing = dir.path().join("missing");
let sink = HealHints::default();
sink.set_enabled(true);
let (read_before, decoded_before) = (metrics.bytes_read(), metrics.bytes_decoded());
let scheduled = maybe_record_persistent_heal(
table.global_id(),
&missing,
&table.file_accessor,
&table.regions.tli,
BlockType::Index,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
Some(&sink),
&metrics,
ReadCharge::Foreground,
);
assert!(!scheduled, "an unopenable file confirms nothing");
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
(read_before, decoded_before),
"no re-read was issued, so none may be counted",
);
Ok(())
}
#[cfg(feature = "metrics")]
fn recover_counted(
file: &std::path::Path,
checksum: Checksum,
) -> crate::Result<(Table, Arc<crate::metrics::Metrics>)> {
let metrics = Arc::new(crate::metrics::Metrics::default());
let mut params = test_recover_params(file.to_path_buf(), checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
params.metrics = metrics.clone();
Ok((Table::recover(params)?, metrics))
}
#[cfg(all(feature = "metrics", feature = "std"))]
#[test]
fn a_repair_restriction_bound_counts_no_bytes() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0..50u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let (table, metrics) = recover_counted(&file, checksum)?;
let geometry = PunchGeometry {
verdict: PunchProbe::Punched,
first_readable: None,
after_last_zeroed: Some(0),
irregular: false,
};
let before = (metrics.bytes_read(), metrics.bytes_decoded());
let bound = table.greedy_restriction_bound(&geometry)?;
assert!(bound.is_some(), "the walk anchors on the first block's key");
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
before,
"a repair walk is maintenance, not a caller's read",
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "columnar"))]
#[test]
fn salvage_reads_of_a_columnar_table_count_no_bytes() -> crate::Result<()> {
use crate::config::DeleteStrategy;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.delete_strategy(DeleteStrategy::MergeOnRead);
for i in 0..64u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
for pos in [5u32, 20, 40] {
writer.delete_bitmap_mut().insert(pos);
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let (table, metrics) = recover_counted(&file, checksum)?;
let before = (metrics.bytes_read(), metrics.bytes_decoded());
assert!(
table.delete_positions_verified()?,
"the positions check out"
);
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
before,
"verifying delete positions is maintenance, not a caller's read",
);
let (table, metrics) = recover_counted(&file, checksum)?;
let first = table
.data_block_handles()
.next()
.expect("the table has a data block")?;
let handle = *first.as_ref();
let before = (metrics.bytes_read(), metrics.bytes_decoded());
assert!(
table.load_columnar_block_masked(&handle)?.is_some(),
"the block keeps live rows",
);
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
before,
"re-emitting a masked block is maintenance, not a caller's read",
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn a_handle_refused_before_reading_counts_no_bytes() -> crate::Result<()> {
use crate::{
CompressionType,
cache::Cache,
table::{BlockHandle, block::BlockType, util::load_block},
};
let dir = tempdir()?;
let (table, metrics, _frame) = one_row_table_and_its_index_frame(&dir)?;
let read_before = metrics.bytes_read();
let result = load_block(
table.global_id(),
&table.path,
&table.file_accessor,
&Cache::with_capacity_bytes(10_000_000),
&BlockHandle::new(table.regions.tli.offset(), u32::MAX),
BlockType::Data,
CompressionType::None,
None,
None,
#[cfg(zstd_any)]
None,
None,
&metrics,
crate::table::util::ReadCharge::Foreground,
);
assert!(
matches!(&result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
"a handle past the size cap must be refused before reading",
);
assert_eq!(
metrics.bytes_read(),
read_before,
"no read was issued, so none may be counted",
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn a_staged_block_rejected_for_its_role_counts_what_its_transform_decoded() -> crate::Result<()> {
use crate::table::block::BlockType;
let dir = tempdir()?;
let (table, metrics, frame) = one_row_table_and_its_index_frame(&dir)?;
let decoded_before = metrics.bytes_decoded();
let result =
table.decode_block_from_bytes(&frame, table.regions.tli.offset().0, BlockType::Filter);
assert!(
matches!(&result, Err(crate::Error::InvalidTag(("BlockType", _)))),
"the index block must be refused as a filter block",
);
assert_eq!(
metrics.bytes_decoded() - decoded_before,
index_frame_decoded_len(&table)?,
"the transform ran before the role check, so its output must be counted",
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn a_chunk_decoded_block_rejected_for_its_role_counts_what_its_transform_decoded()
-> crate::Result<()> {
let dir = tempdir()?;
let (table, metrics, frame) = one_row_table_and_its_index_frame(&dir)?;
let decoded_before = metrics.bytes_decoded();
let result = table.decode_data_block_keeping(&frame, &table.regions.tli, &mut 0, None);
assert!(
matches!(&result, Err(crate::Error::InvalidTag(("BlockType", _)))),
"the index block must be refused as a data block",
);
assert_eq!(
metrics.bytes_decoded() - decoded_before,
index_frame_decoded_len(&table)?,
"the transform ran before the role check, so its output must be counted",
);
Ok(())
}
#[cfg(all(feature = "metrics", feature = "page_ecc", feature = "std"))]
#[test]
fn a_patrol_scrub_that_corrects_a_block_counts_no_bytes() -> crate::Result<()> {
use crate::{
Cache,
heal_hints::HealHints,
table::{
BlockHandle,
block::{BlockType, EccParams, Header},
util::{BlockScrubOutcome, scrub_block},
},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer =
Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_ecc(Some(EccParams::RS_4_2));
for i in 0..200u32 {
let key = format!("key{i:05}");
writer.write(InternalValue::from_components(
key.as_bytes(),
b"value-payload-bytes",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
params.metrics = metrics.clone();
Table::recover(params)?
};
let table_id = table.global_id();
let keyed = table
.block_index
.iter()
.next()
.expect("the table has a data block")?;
let handle = BlockHandle::new(keyed.offset(), keyed.size());
let mut bytes = std::fs::read(&file)?;
let pos =
usize::try_from(handle.offset().0).expect("block offset fits usize") + Header::MIN_LEN + 3;
bytes[pos] ^= 0x80;
std::fs::write(&file, &bytes)?;
table.file_accessor.remove_for_table(&table_id);
let sink = HealHints::default();
sink.set_enabled(true);
let (read_before, decoded_before) = (metrics.bytes_read(), metrics.bytes_decoded());
let outcome = scrub_block(
table_id,
&table.path,
&table.file_accessor,
&handle,
BlockType::Data,
table.metadata.data_block_compression,
None,
table.metadata.ecc_params,
#[cfg(zstd_any)]
None,
Some(&sink),
&metrics,
)?;
assert_eq!(
outcome,
BlockScrubOutcome::Corrected { scheduled: true },
"the scrub corrects the block and confirms the fault persists",
);
assert_eq!(
(metrics.bytes_read(), metrics.bytes_decoded()),
(read_before, decoded_before),
"neither the scrub's read nor its confirming re-read is a foreground read",
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn a_patrol_scrub_leaves_the_block_and_descriptor_caches_as_it_found_them() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0..200u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value-payload-bytes",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
let descriptors = Arc::new(DescriptorTable::new(10));
let mut params = test_recover_params(file, checksum);
params.cache = cache.clone();
params.descriptor_table = Some(descriptors.clone());
let table = Table::recover(params)?;
table.file_accessor.remove_for_table(&table.global_id());
assert_eq!(cache.size(), 0, "precondition: the index is cold");
assert_eq!(descriptors.len(), 0, "precondition: the descriptor is cold");
let report = table.scrub_data_blocks();
assert!(report.errors.is_empty(), "the clean table scrubs clean");
assert!(report.blocks_scanned > 1, "the scrub walked every block");
assert_eq!(
cache.size(),
0,
"the scrub's index walk must not fill the cache"
);
assert_eq!(
descriptors.len(),
0,
"the scrub must not cache the descriptor it opened",
);
Ok(())
}
#[cfg(all(feature = "page_ecc", feature = "std"))]
#[test]
fn an_untraced_read_confirming_an_ecc_correction_leaves_the_descriptor_cache_alone()
-> crate::Result<()> {
use crate::{
heal_hints::HealHints,
table::{
BlockHandle,
block::{BlockType, EccParams, Header},
util::{ReadCharge, load_block},
},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer =
Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_ecc(Some(EccParams::RS_4_2));
for i in 0..200u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value-payload-bytes",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let descriptors = Arc::new(DescriptorTable::new(10));
let mut params = test_recover_params(file.clone(), checksum);
params.descriptor_table = Some(descriptors.clone());
#[cfg(feature = "metrics")]
let metrics = params.metrics.clone();
let table = Table::recover(params)?;
let table_id = table.global_id();
let keyed = table
.block_index
.iter()
.next()
.expect("the table has a data block")?;
let handle = BlockHandle::new(keyed.offset(), keyed.size());
let mut bytes = std::fs::read(&file)?;
let pos =
usize::try_from(handle.offset().0).expect("block offset fits usize") + Header::MIN_LEN + 3;
bytes[pos] ^= 0x80;
std::fs::write(&file, &bytes)?;
table.file_accessor.remove_for_table(&table_id);
assert_eq!(descriptors.len(), 0, "precondition: the descriptor is cold");
let sink = HealHints::default();
sink.set_enabled(true);
load_block(
table_id,
&table.path,
&table.file_accessor,
&Cache::with_capacity_bytes(10_000_000),
&handle,
BlockType::Data,
table.metadata.data_block_compression,
None,
table.metadata.ecc_params,
#[cfg(zstd_any)]
None,
Some(&sink),
#[cfg(feature = "metrics")]
&metrics,
ReadCharge::Untraced,
)?;
assert_eq!(
sink.snapshot(),
vec![table_id],
"the confirmed fault is still queued for healing",
);
assert_eq!(
descriptors.len(),
0,
"neither the read nor its confirming re-read may cache the descriptor",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn load_block_records_heal_hint_on_persistent_ecc_correction() -> crate::Result<()> {
use crate::{
Cache, InternalValue,
fs::StdFs,
heal_hints::HealHints,
table::{
BlockHandle,
block::{BlockType, EccParams, Header},
util::load_block,
},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let scheme = EccParams::RS_4_2;
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_ecc(Some(scheme));
for i in 0..200u32 {
let key = format!("key{i:05}");
writer.write(InternalValue::from_components(
key.as_bytes(),
b"value-payload-bytes",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing items"
)]
let (_, checksum) = writer.finish()?.unwrap();
#[cfg(feature = "metrics")]
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
#[cfg(feature = "metrics")]
{
params.metrics = metrics.clone();
}
Table::recover(params)?
};
let table_id = table.global_id();
let compression = table.metadata.data_block_compression;
#[expect(clippy::unwrap_used, reason = "table has at least one data block")]
let keyed = table.block_index.iter().next().unwrap()?;
let handle = BlockHandle::new(keyed.offset(), keyed.size());
{
let clean_sink = HealHints::default();
clean_sink.set_enabled(true);
let fresh_cache = Cache::with_capacity_bytes(10_000_000);
let _block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&handle,
BlockType::Data,
compression,
None,
table.metadata.ecc_params,
#[cfg(zstd_any)]
None,
Some(&clean_sink),
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
assert!(
clean_sink.snapshot().is_empty(),
"a clean read must not record a heal hint",
);
}
let mut bytes = std::fs::read(&file)?;
#[allow(
clippy::cast_possible_truncation,
reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
let pos = handle.offset().0 as usize + Header::MIN_LEN + 3;
bytes[pos] ^= 0x80;
std::fs::write(&file, &bytes)?;
table.file_accessor.remove_for_table(&table_id);
let sink = HealHints::default();
sink.set_enabled(true);
let fresh_cache = Cache::with_capacity_bytes(10_000_000);
#[cfg(feature = "metrics")]
let (read_before, decoded_before) = (metrics.bytes_read(), metrics.bytes_decoded());
let block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&handle,
BlockType::Data,
compression,
None,
table.metadata.ecc_params,
#[cfg(zstd_any)]
None,
Some(&sink),
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
assert_eq!(
block.header.block_type,
BlockType::Data,
"repaired read still yields a valid data block",
);
assert_eq!(
sink.snapshot(),
vec![table_id],
"a persistent ECC correction must queue the SST for healing",
);
#[cfg(feature = "metrics")]
{
let size = u64::from(handle.size());
assert_eq!(
metrics.bytes_read() - read_before,
2 * size,
"the corrected read and its confirming re-read were both asked of the filesystem",
);
assert_eq!(
metrics.bytes_decoded() - decoded_before,
2 * block.data.len() as u64,
"the confirming re-read decoded the block a second time",
);
}
table.file_accessor.remove_for_table(&table_id);
let off_sink = HealHints::default(); let fresh_cache = Cache::with_capacity_bytes(10_000_000);
#[cfg(feature = "metrics")]
let read_before = metrics.bytes_read();
let block = load_block(
table_id,
&table.path,
&table.file_accessor,
&fresh_cache,
&handle,
BlockType::Data,
compression,
None,
table.metadata.ecc_params,
#[cfg(zstd_any)]
None,
Some(&off_sink),
#[cfg(feature = "metrics")]
&metrics,
crate::table::util::ReadCharge::Foreground,
)?;
assert_eq!(
block.header.block_type,
BlockType::Data,
"disabled auto-heal still returns repaired data",
);
assert!(
off_sink.snapshot().is_empty(),
"auto_heal off must not schedule a rewrite",
);
#[cfg(feature = "metrics")]
assert_eq!(
metrics.bytes_read() - read_before,
u64::from(handle.size()),
"with auto-heal off there is no confirming re-read to count",
);
Ok(())
}
#[cfg(feature = "metrics")]
#[test]
fn excised_probe_counts_the_bytes_it_reads() -> crate::Result<()> {
use crate::{
Cache, InternalValue,
fs::StdFs,
table::{BlockHandle, block_index::BlockIndex as _},
};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for i in 0..200u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").as_bytes(),
b"value-payload-bytes",
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing items"
)]
let (_, checksum) = writer.finish()?.unwrap();
let metrics = Arc::new(crate::metrics::Metrics::default());
let table = {
let mut params = test_recover_params(file.clone(), checksum);
params.cache = Arc::new(Cache::with_capacity_bytes(10_000_000));
params.metrics = metrics.clone();
Table::recover(params)?
};
#[expect(clippy::unwrap_used, reason = "table has at least one data block")]
let keyed = table.block_index.iter().next().unwrap()?;
let handle = BlockHandle::new(keyed.offset(), keyed.size());
let mut bytes = std::fs::read(&file)?;
let start = usize::try_from(handle.offset().0).expect("block offset fits usize");
let len = usize::try_from(handle.size()).expect("block size fits usize");
bytes[start..start + len].fill(0);
std::fs::write(&file, &bytes)?;
table.file_accessor.remove_for_table(&table.global_id());
let before = metrics.bytes_read();
let err = table
.load_data_block(&handle, &crate::table::util::LookupFd::default())
.err();
assert!(
matches!(err, Some(crate::Error::Excised { .. })),
"a zeroed extent must read as excised, got {err:?}",
);
let size = u64::from(handle.size());
assert_eq!(
metrics.bytes_read() - before,
size + size.min(64) + size,
"the load, the 64-byte probe and the full-extent probe were all asked of the filesystem",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
fn build_ecc_sst_for_heal(
file: &std::path::Path,
fs: Arc<dyn crate::fs::Fs>,
scheme: crate::table::block::EccParams,
n: u32,
) -> crate::Checksum {
#[expect(
clippy::expect_used,
reason = "test setup; a panic is the failure signal"
)]
let mut writer = Writer::new(file.to_path_buf(), 0, 0, fs)
.expect("open writer")
.use_data_block_size(256)
.use_ecc(Some(scheme));
for i in 0..n {
#[expect(clippy::expect_used, reason = "test setup")]
writer
.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value-payload-bytes".to_vec(),
u64::from(i) + 1,
crate::ValueType::Value,
))
.expect("write");
}
#[expect(clippy::expect_used, reason = "finish() returns Some after writes")]
let (_, checksum) = writer.finish().expect("finish").expect("non-empty");
checksum
}
#[cfg(feature = "page_ecc")]
#[expect(
clippy::expect_used,
reason = "test setup; a panic is the failure signal"
)]
fn recover_table_on(
file: &std::path::Path,
checksum: crate::Checksum,
fs: Arc<dyn crate::fs::Fs>,
) -> Table {
let mut params = test_recover_params(file.to_path_buf(), checksum);
params.cache = Arc::new(crate::Cache::with_capacity_bytes(10_000_000));
params.fs = fs;
Table::recover(params).expect("recover table")
}
#[cfg(feature = "page_ecc")]
fn first_data_block_offset(table: &Table) -> u64 {
use crate::table::block_index::BlockIndex as _;
let Some(keyed) = table.block_index.iter().find_map(Result::ok) else {
panic!("a non-empty SST has at least one data block");
};
keyed.offset().0
}
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
clippy::cast_possible_truncation,
reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_restores_a_secded_block_byte_for_byte() -> crate::Result<()> {
use crate::table::block::{EccParams, Header};
let dir = tempdir()?;
let file = dir.path().join("table");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::Secded, 200);
let original = std::fs::read(&file)?;
let first_off =
first_data_block_offset(&recover_table_on(&file, checksum, Arc::clone(&fs))) as usize;
let pos = first_off + Header::MIN_LEN + 3;
let mut bytes = original.clone();
if let Some(b) = bytes.get_mut(pos) {
*b ^= 0x01;
}
std::fs::write(&file, &bytes)?;
assert_ne!(bytes, original, "the seeded fault changed the file");
let table = recover_table_on(&file, checksum, Arc::clone(&fs));
let (report, attributable) =
table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
assert_eq!(report.blocks_healed_in_place, 1, "{report:?}");
assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
assert!(
!attributable,
"a pre-heal digest differing from the manifest must not attribute",
);
let healed = std::fs::read(&file)?;
assert_eq!(
healed, original,
"SEC-DED in-place heal restores the block byte-for-byte",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
clippy::cast_possible_truncation,
reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_attributes_a_matching_pre_heal_digest() -> crate::Result<()> {
use crate::coding::Decode;
use crate::table::block::{EccParams, Header};
let dir = tempdir()?;
let file = dir.path().join("table");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
let healthy = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
let first_off =
first_data_block_offset(&recover_table_on(&file, healthy, Arc::clone(&fs))) as usize;
let mut bytes = std::fs::read(&file)?;
let Some(mut cursor) = bytes.get(first_off..) else {
panic!("first data block within the file");
};
let header = Header::decode_from(&mut cursor)?;
let trailer_pos =
first_off + Header::header_len(header.block_type) + header.data_length as usize;
let Some(slot) = bytes.get_mut(trailer_pos) else {
panic!("parity trailer within the file");
};
*slot ^= 0xFF;
std::fs::write(&file, &bytes)?;
let rotted = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &file)?);
let table = recover_table_on(&file, rotted, Arc::clone(&fs));
let (report, attributable) =
table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
assert_eq!(
report.blocks_healed_in_place, 1,
"the rotted trailer is rebuilt in place: {report:?}",
);
assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
assert!(
attributable,
"a pre-heal digest matching the manifest attributes the mismatch to \
this pass's own verified corrections",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
clippy::cast_possible_truncation,
reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn predict_heal_streams_the_same_digest_as_materializing_corrections() -> crate::Result<()> {
use crate::coding::Decode;
use crate::table::block::{EccParams, Header};
let dir = tempdir()?;
let file = dir.path().join("table");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
let healthy = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
let first_off =
first_data_block_offset(&recover_table_on(&file, healthy, Arc::clone(&fs))) as usize;
let mut bytes = std::fs::read(&file)?;
let Some(mut cursor) = bytes.get(first_off..) else {
panic!("first data block within the file");
};
let header = Header::decode_from(&mut cursor)?;
let trailer_pos =
first_off + Header::header_len(header.block_type) + header.data_length as usize;
let Some(slot) = bytes.get_mut(trailer_pos) else {
panic!("parity trailer within the file");
};
*slot ^= 0xFF;
std::fs::write(&file, &bytes)?;
let rotted = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &file)?);
let table = recover_table_on(&file, rotted, Arc::clone(&fs));
let transform = crate::table::util::build_block_transform(
table.metadata.data_block_compression,
table.encryption.as_deref(),
table.metadata.ecc_params,
#[cfg(zstd_any)]
table.zstd_dictionary.as_deref(),
)?;
let fh = fs.open(&file, &crate::fs::FsOpenOptions::new().read(true))?;
let (streamed, offsets) = table.predict_heal_digest_and_offsets(fh.as_ref(), &transform, 0)?;
let mut corrections: Vec<(u64, Vec<u8>)> = Vec::new();
for entry in table.block_index.iter() {
let keyed = entry?;
let entry_handle = *keyed.as_ref();
for (handle, role) in table.data_unit_blocks(&entry_handle)? {
if let Some(c) =
table.heal_correction_for_block(fh.as_ref(), handle, role, &transform)?
{
corrections.push(c);
}
}
}
let materialized =
crate::repair::compute_table_checksum_with_overrides(&*fs, &file, 0, &corrections)?;
assert_eq!(
streamed, materialized,
"the streamed digest must equal the materialized-overrides digest",
);
assert_eq!(
offsets.len(),
corrections.len(),
"one predicted offset per correction: {corrections:?}",
);
for (off, _) in &corrections {
assert!(
offsets.contains(off),
"every correction offset ({off}) is in the predicted set",
);
}
assert_eq!(corrections.len(), 1, "exactly one block was rotted");
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
#[allow(
clippy::cast_possible_truncation,
reason = "in-file block offset fits usize; only narrows on 32-bit targets"
)]
fn heal_data_blocks_in_place_reports_a_block_whose_write_back_fails() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
use crate::io::ErrorKind;
use crate::table::block::{EccParams, Header};
let dir = tempdir()?;
let file = dir.path().join("table");
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
let fs: Arc<dyn crate::fs::Fs> = Arc::new(fault);
let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
let first_off =
first_data_block_offset(&recover_table_on(&file, checksum, Arc::clone(&fs))) as usize;
let pos = first_off + Header::MIN_LEN + 3;
let mut bytes = std::fs::read(&file)?;
if let Some(b) = bytes.get_mut(pos) {
*b ^= 0x80;
}
std::fs::write(&file, &bytes)?;
injector.arm(FaultRule::new(FaultOp::Write, Fault::Error(ErrorKind::Other)).skip(1));
let table = recover_table_on(&file, checksum, fs);
let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
assert_eq!(
report.blocks_healed_in_place, 0,
"a failed write-back heals nothing: {report:?}",
);
assert!(
report.uncorrectable_blocks >= 1,
"the failed write-back is reported, not silently dropped: {report:?}",
);
assert!(
report
.errors
.iter()
.any(|e| matches!(e, crate::scrub::ScrubError::UncorrectableBlock { .. })),
"the finding is an UncorrectableBlock: {report:?}",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn heal_data_blocks_in_place_is_a_noop_on_a_non_ecc_sst() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(crate::fs::StdFs);
let mut writer = Writer::new(file.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
for i in 0..200u32 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"value-payload-bytes".to_vec(),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let Some((_, checksum)) = writer.finish()? else {
panic!("non-empty SST");
};
let table = recover_table_on(&file, checksum, fs);
let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
assert!(
report.blocks_scanned > 0,
"the walk inspected blocks: {report:?}"
);
assert_eq!(
report.blocks_healed_in_place, 0,
"no parity means nothing to heal: {report:?}",
);
assert_eq!(report.uncorrectable_blocks, 0, "{report:?}");
assert!(report.errors.is_empty(), "{report:?}");
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn heal_data_blocks_in_place_reports_when_the_file_cannot_be_opened() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
use crate::io::ErrorKind;
use crate::table::block::{EccParams, Header};
let dir = tempdir()?;
let file = dir.path().join("table");
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
let fs: Arc<dyn crate::fs::Fs> = Arc::new(fault);
let checksum = build_ecc_sst_for_heal(&file, Arc::clone(&fs), EccParams::RS_4_2, 200);
let table = recover_table_on(&file, checksum, Arc::clone(&fs));
{
use crate::table::block_index::BlockIndex as _;
let Some(keyed) = table.block_index.iter().find_map(Result::ok) else {
panic!("the SST has at least one data block");
};
let base = usize::try_from(keyed.offset().0).unwrap_or(usize::MAX);
let mut bytes = std::fs::read(&file)?;
let Some(payload) = bytes.get_mut(base + Header::MIN_LEN..base + keyed.size() as usize)
else {
panic!("block payload range within the file");
};
for b in payload {
*b ^= 0xFF;
}
std::fs::write(&file, &bytes)?;
}
injector.arm(FaultRule::new(FaultOp::Open, Fault::Error(ErrorKind::Other)).once());
let (report, _) = table.heal_data_blocks_in_place(crate::fs::SyncMode::Full, table.checksum());
assert!(
report.blocks_scanned >= 1,
"the read-only fallback still scans the table: {report:?}",
);
assert_eq!(
report.blocks_healed_in_place, 0,
"nothing is healed without a writable file: {report:?}",
);
assert!(
report.uncorrectable_blocks >= 1,
"the fallback scrub reports the seeded corruption: {report:?}",
);
assert!(!report.is_ok(), "corruption fails the pass: {report:?}");
assert!(
report
.errors
.iter()
.any(|e| matches!(e, crate::scrub::ScrubError::BlockIndexUnreadable { .. })),
"the failed read+write open is still reported: {report:?}",
);
Ok(())
}
#[test]
#[expect(
clippy::expect_used,
reason = "test invariants: key and value patterns must exist in the meta block"
)]
#[expect(
clippy::indexing_slicing,
reason = "test fixture: deliberate slice operations on controlled meta block bytes"
)]
fn meta_seqno_kv_max_corruption_returns_invalid_data() -> crate::Result<()> {
use super::block::Header;
use super::meta::ParsedMeta;
use super::regions::ParsedRegions;
use crate::coding::{Decode, Encode};
use std::io::{Seek, Write};
let dir = tempfile::tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for (i, key) in (b'a'..=b'e').enumerate() {
writer.write(InternalValue::from_components(
[key],
b"val",
(i as u64) + 1,
crate::ValueType::Value,
))?;
}
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing data items"
)]
let _ = writer.finish()?.unwrap();
{
let mut f = std::fs::File::open(&file)?;
let trailer = crate::sfa::Reader::from_reader(&mut f)?;
let regions = ParsedRegions::parse_from_toc(trailer.toc())?;
let meta_handle = regions.metadata;
let raw_block =
crate::file::read_exact(&f, *meta_handle.offset(), meta_handle.size() as usize)?;
let header_len = Header::header_len(crate::table::block::BlockType::Meta);
let payload = &raw_block[header_len..];
let needle = b"seqno#kv_max";
let key_pos = payload
.windows(needle.len())
.position(|w| w == needle)
.expect("seqno#kv_max key must be present in the meta block payload");
let search_start = key_pos + needle.len();
let original_le = 5u64.to_le_bytes();
let val_rel = payload[search_start..]
.windows(original_le.len())
.position(|w| w == original_le)
.expect("original LE value must appear after the key");
let val_offset_in_payload = search_start + val_rel;
let mut tampered_payload = payload.to_vec();
tampered_payload[val_offset_in_payload..val_offset_in_payload + 8]
.copy_from_slice(&u64::MAX.to_le_bytes());
let mut orig_header = Header::decode_from(&mut &raw_block[..header_len])?;
orig_header.bind_checksum(
crate::Checksum::from_raw(crate::hash::hash128(&tampered_payload)),
crate::table::block::ChecksumAt::block(
0,
crate::table::block::BlockType::Meta,
*meta_handle.offset(),
),
);
let new_header = orig_header.encode_into_vec();
let mut wf = std::fs::OpenOptions::new().write(true).open(&file)?;
wf.seek(std::io::SeekFrom::Start(*meta_handle.offset()))?;
wf.write_all(&new_header)?;
wf.write_all(&tampered_payload)?;
wf.sync_all()?;
}
{
let mut f = std::fs::File::open(&file)?;
let trailer = crate::sfa::Reader::from_reader(&mut f)?;
let regions = ParsedRegions::parse_from_toc(trailer.toc())?;
let result = ParsedMeta::load_with_handle(&f, ®ions.metadata, None, None);
let err = result.expect_err("corrupted seqno#kv_max should cause an error");
assert!(
matches!(&err, crate::Error::Io(e) if e.kind() == crate::io::ErrorKind::InvalidData),
"expected InvalidData, got: {err:?}",
);
}
Ok(())
}
#[test]
fn meta_mid_and_tail_have_identical_created_at() -> crate::Result<()> {
use super::meta::ParsedMeta;
use super::regions::ParsedRegions;
let dir = tempfile::tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for (i, key) in (b'a'..=b'e').enumerate() {
writer.write(InternalValue::from_components(
[key],
b"val",
(i as u64) + 1,
crate::ValueType::Value,
))?;
}
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing data items"
)]
let _ = writer.finish()?.unwrap();
let mut f = std::fs::File::open(&file)?;
let trailer = crate::sfa::Reader::from_reader(&mut f)?;
let regions = ParsedRegions::parse_from_toc(trailer.toc())?;
let tail = ParsedMeta::load_with_handle(&f, ®ions.metadata, None, None)?;
let mid_handle = regions
.metadata_mid
.expect("writer must emit meta_mid alongside meta");
let mid = ParsedMeta::load_with_handle(&f, &mid_handle, None, None)?;
assert_eq!(
tail.created_at, mid.created_at,
"MID and TAIL meta copies must share an identical created_at \
(writer must snapshot the timestamp once and pass it to both \
write_meta_section calls; observed tail={:?} mid={:?})",
tail.created_at, mid.created_at,
);
Ok(())
}
#[test]
fn meta_mid_and_tail_have_identical_file_size() -> crate::Result<()> {
use super::meta::ParsedMeta;
use super::regions::ParsedRegions;
let dir = tempfile::tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
for (i, key) in (b'a'..=b'e').enumerate() {
writer.write(InternalValue::from_components(
[key],
b"val",
(i as u64) + 1,
crate::ValueType::Value,
))?;
}
#[expect(
clippy::unwrap_used,
reason = "finish() returns Some after writing data items"
)]
let _ = writer.finish()?.unwrap();
let mut f = std::fs::File::open(&file)?;
let trailer = crate::sfa::Reader::from_reader(&mut f)?;
let regions = ParsedRegions::parse_from_toc(trailer.toc())?;
let tail = ParsedMeta::load_with_handle(&f, ®ions.metadata, None, None)?;
let mid_handle = regions
.metadata_mid
.expect("writer must emit meta_mid alongside meta");
let mid = ParsedMeta::load_with_handle(&f, &mid_handle, None, None)?;
assert_eq!(
tail.file_size, mid.file_size,
"MID and TAIL meta copies must store an identical file_size \
(both observe the same `self.meta.file_pos` because no \
post-data section bumps it); observed tail={} mid={}",
tail.file_size, mid.file_size,
);
assert_ne!(
mid.file_size, 0,
"MID file_size must not be the legacy 0 sentinel — that pushed \
the recovery path through std::fs::metadata, which bypasses \
the pluggable Fs backend"
);
Ok(())
}
#[test]
fn bloom_may_contain_key_full_filter() -> crate::Result<()> {
let items: Vec<InternalValue> = ["a", "c", "e"]
.iter()
.enumerate()
.map(|(i, &k)| {
InternalValue::from_components(k, "v", i as u64 + 1, crate::ValueType::Value)
})
.collect();
test_with_table(
&items,
|table| {
let hash_a = hash64(b"a");
let hash_b = hash64(b"b");
assert!(
table.bloom_may_contain_key(b"a", hash_a)?,
"bloom_may_contain_key must not reject existing key"
);
assert!(
table.bloom_may_contain_key_hash(hash_a)?,
"bloom_may_contain_key_hash must not reject existing key"
);
let key_result = table.bloom_may_contain_key(b"b", hash_b)?;
let hash_result = table.bloom_may_contain_key_hash(hash_b)?;
assert_eq!(
key_result, hash_result,
"full filter: key-based and hash-only should agree"
);
Ok(())
},
None,
Some(|w: Writer| w.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(10.0))),
)
}
#[test]
fn bloom_may_contain_key_partitioned_filter() -> crate::Result<()> {
let items: Vec<InternalValue> = (0u64..100)
.map(|i| {
let key = format!("key_{i:04}");
InternalValue::from_components(key, "v", i + 1, crate::ValueType::Value)
})
.collect();
test_with_table(
&items,
|table| {
let hash_exist = hash64(b"key_0050");
assert!(
table.bloom_may_contain_key(b"key_0050", hash_exist)?,
"bloom must not reject existing key in partitioned filter"
);
let hash_beyond = hash64(b"zzz_beyond");
assert!(
!table.bloom_may_contain_key(b"zzz_beyond", hash_beyond)?,
"key beyond all partitions should be rejected when partition index is available"
);
assert!(
table.bloom_may_contain_key_hash(hash_beyond)?,
"hash-only bloom check should remain conservative for partitioned filters"
);
Ok(())
},
None,
Some(|w: Writer| {
w.use_bloom_policy(BloomConstructionPolicy::BitsPerKey(10.0))
.use_partitioned_filter()
}),
)
}
#[test]
fn two_level_index_scan_skips_empty_child_partition() -> crate::Result<()> {
use crate::ValueType::Value;
use crate::table::block_index::{BlockIndex, BlockIndexIter};
let items: Vec<InternalValue> = ["a", "b", "c", "d", "e", "f", "g", "h"]
.iter()
.enumerate()
.map(|(i, k)| InternalValue::from_components(*k, format!("v{i}"), (i + 1) as u64, Value))
.collect();
let dir = tempfile::tempdir()?;
let file = dir.path().join("two_level_skip");
let mut writer = crate::table::Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_partitioned_index()
.use_data_block_size(1)
.use_meta_partition_size(3);
for item in items.iter().cloned() {
writer.write(item)?;
}
writer.finish()?;
let table = {
let mut params = test_recover_params(file, crate::Checksum::from_raw(0));
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
params.pin_filter = true;
crate::Table::recover(params)?
};
assert!(
table.regions.index.is_some(),
"table must use partitioned (two-level) index",
);
assert!(
table.metadata.index_block_count > 1,
"table must have >1 index partitions, got {}",
table.metadata.index_block_count,
);
let all_handles: Vec<_> = {
let it = table.block_index.iter();
it.collect::<Result<Vec<_>, _>>()?
};
assert_eq!(
all_handles.len(),
items.len(),
"full scan should yield one block handle per data block",
);
{
let mut it = table.block_index.iter();
assert!(it.seek_lower(b"d", u64::MAX));
let forward_keys: Vec<_> = it
.map(|r| r.map(|h| h.end_key().to_vec()))
.collect::<Result<Vec<_>, _>>()?;
assert_eq!(
forward_keys,
vec![
b"d".to_vec(),
b"e".to_vec(),
b"f".to_vec(),
b"g".to_vec(),
b"h".to_vec(),
],
"forward scan from 'd' should yield exactly d..h",
);
}
{
let mut it = table.block_index.iter();
assert!(it.seek_upper(b"e", 0));
let mut backward_keys = Vec::new();
while let Some(res) = it.next_back() {
backward_keys.push(res?.end_key().to_vec());
}
assert_eq!(
backward_keys,
vec![
b"f".to_vec(),
b"e".to_vec(),
b"d".to_vec(),
b"c".to_vec(),
b"b".to_vec(),
b"a".to_vec(),
],
"backward scan up to 'e' should yield f..a in reverse",
);
}
{
let mut it = table.block_index.iter();
assert!(it.seek_lower(b"c", u64::MAX));
assert!(it.seek_upper(b"f", 0));
let mut forward_keys = vec![];
let mut backward_keys = vec![];
if let Some(res) = it.next() {
forward_keys.push(res?.end_key().to_vec());
}
if let Some(res) = it.next() {
forward_keys.push(res?.end_key().to_vec());
}
while let Some(res) = it.next_back() {
backward_keys.push(res?.end_key().to_vec());
}
assert_eq!(forward_keys, vec![b"c".to_vec(), b"d".to_vec()]);
assert_eq!(
backward_keys,
vec![b"g".to_vec(), b"f".to_vec(), b"e".to_vec()]
);
assert!(it.next().is_none(), "iterator should be exhausted");
}
Ok(())
}
#[test]
fn batch_get_empty_input_returns_empty_results() -> crate::Result<()> {
let items = [crate::InternalValue::from_components(
b"a",
b"v",
0,
crate::ValueType::Value,
)];
test_with_table(
&items,
|table| {
let r = table.batch_get(&[], SeqNo::MAX)?;
assert!(r.is_empty(), "empty input must yield empty result vec");
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn batch_get_counts_each_key_it_reads_as_a_point_read() -> crate::Result<()> {
let items: Vec<_> = ["a", "b", "c"]
.iter()
.map(|k| {
crate::InternalValue::from_components(k.as_bytes(), b"v", 0, crate::ValueType::Value)
})
.collect();
test_with_table(
&items,
|table| {
let before = table.point_reads();
let batch: Vec<(&[u8], u64)> = vec![
(b"a", hash64(b"a")),
(b"b", hash64(b"b")),
(b"c", hash64(b"c")),
];
table.batch_get(&batch, SeqNo::MAX)?;
assert_eq!(table.point_reads() - before, 3);
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_single_block_multiple_keys_returns_in_input_order() -> crate::Result<()> {
let items: Vec<_> = ["a", "b", "c"]
.iter()
.enumerate()
.map(|(i, k)| {
crate::InternalValue::from_components(
k.as_bytes(),
format!("val-{k}").as_bytes(),
u64::try_from(i).expect("test fixture index fits in u64"),
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let batch: Vec<(&[u8], u64)> = vec![
(b"a", hash64(b"a")),
(b"b", hash64(b"b")),
(b"c", hash64(b"c")),
];
let results = table.batch_get(&batch, SeqNo::MAX)?;
assert_eq!(results.len(), 3, "one result slot per input key");
assert_eq!(&*results[0].as_ref().unwrap().value, b"val-a");
assert_eq!(&*results[1].as_ref().unwrap().value, b"val-b");
assert_eq!(&*results[2].as_ref().unwrap().value, b"val-c");
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_keys_spread_across_blocks_return_correct_values() -> crate::Result<()> {
let items: Vec<_> = (0u32..8)
.map(|i| {
let key = format!("key-{i:04}");
let value = format!("val-{i:04}");
crate::InternalValue::from_components(
key.as_bytes(),
value.as_bytes(),
u64::from(i),
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let queries: Vec<(&[u8], u64)> = vec![
(b"key-0000" as &[u8], hash64(b"key-0000")),
(b"key-0002" as &[u8], hash64(b"key-0002")),
(b"key-0005" as &[u8], hash64(b"key-0005")),
(b"key-0007" as &[u8], hash64(b"key-0007")),
];
let results = table.batch_get(&queries, SeqNo::MAX)?;
assert_eq!(results.len(), 4);
assert_eq!(&*results[0].as_ref().unwrap().value, b"val-0000");
assert_eq!(&*results[1].as_ref().unwrap().value, b"val-0002");
assert_eq!(&*results[2].as_ref().unwrap().value, b"val-0005");
assert_eq!(&*results[3].as_ref().unwrap().value, b"val-0007");
Ok(())
},
Some(1),
Some(|writer: Writer| writer.use_data_block_size(64)),
)
}
#[test]
#[expect(clippy::unwrap_used)]
fn batch_get_missing_keys_return_none_present_keys_return_some() -> crate::Result<()> {
let items: Vec<_> = ["b", "d", "f"]
.iter()
.enumerate()
.map(|(i, k)| {
crate::InternalValue::from_components(
k.as_bytes(),
format!("val-{k}").as_bytes(),
u64::try_from(i).expect("test fixture index fits in u64"),
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let batch: Vec<(&[u8], u64)> = vec![
(b"a" as &[u8], hash64(b"a")), (b"b" as &[u8], hash64(b"b")), (b"c" as &[u8], hash64(b"c")), (b"d" as &[u8], hash64(b"d")), (b"f" as &[u8], hash64(b"f")), (b"g" as &[u8], hash64(b"g")), ];
let results = table.batch_get(&batch, SeqNo::MAX)?;
assert_eq!(results.len(), 6);
assert!(results[0].is_none(), "key 'a' is absent");
assert_eq!(&*results[1].as_ref().unwrap().value, b"val-b");
assert!(results[2].is_none(), "key 'c' is absent");
assert_eq!(&*results[3].as_ref().unwrap().value, b"val-d");
assert_eq!(&*results[4].as_ref().unwrap().value, b"val-f");
assert!(results[5].is_none(), "key 'g' is absent");
Ok(())
},
None,
Some(|x| x),
)
}
#[test]
fn batch_get_matches_per_key_get() -> crate::Result<()> {
let items: Vec<_> = (0u32..20)
.map(|i| {
let key = format!("k-{i:03}");
let value = format!("v-{i:03}");
crate::InternalValue::from_components(
key.as_bytes(),
value.as_bytes(),
u64::from(i),
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let keys: Vec<Vec<u8>> = (0..25).map(|i| format!("k-{i:03}").into_bytes()).collect();
let batch: Vec<(&[u8], u64)> = keys.iter().map(|k| (k.as_slice(), hash64(k))).collect();
let batch_results = table.batch_get(&batch, SeqNo::MAX)?;
let single_results: Vec<_> = batch
.iter()
.map(|&(k, h)| table.get(k, SeqNo::MAX, h))
.collect::<crate::Result<Vec<_>>>()?;
assert_eq!(batch_results.len(), single_results.len());
for (i, (b, s)) in batch_results.iter().zip(&single_results).enumerate() {
assert_eq!(
b,
s,
"batch/single divergence at index {i} (key={})",
String::from_utf8_lossy(&keys[i]),
);
}
Ok(())
},
Some(2),
Some(|writer: Writer| writer.use_data_block_size(96)),
)
}
#[test]
fn batch_get_same_user_key_across_block_boundary_finds_older_visible_version() -> crate::Result<()>
{
let items = [
crate::InternalValue::from_components(b"0", b"zero", 1, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"5", 5, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"4", 4, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"3", 3, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"2", 2, crate::ValueType::Value),
crate::InternalValue::from_components(b"a", b"1", 1, crate::ValueType::Value),
];
test_with_table(
&items,
|table| {
assert_eq!(2, table.metadata.data_block_count);
let batch: Vec<(&[u8], u64)> = vec![(b"0", hash64(b"0")), (b"a", hash64(b"a"))];
let results = table.batch_get(&batch, 3)?;
assert_eq!(results.len(), 2);
assert_eq!(
&*results[0]
.as_ref()
.expect("0@1 must be found in block 0")
.value,
b"zero",
);
assert_eq!(
&*results[1]
.as_ref()
.expect("a@2 must be found via block 1")
.value,
b"2",
"batch_get must walk past block 0 (end_key=a, but all a-seqnos ≥3) \
into block 1 (end_key=a, seqnos 2 and 1) to find the visible version \
at snapshot 3",
);
let single_zero = table.get(b"0", 3, hash64(b"0"))?;
let single_a = table.get(b"a", 3, hash64(b"a"))?;
assert_eq!(
results[0], single_zero,
"batch_get must match Table::get for '0'"
);
assert_eq!(
results[1], single_a,
"batch_get must match Table::get for 'a'"
);
Ok(())
},
Some(3),
Some(|x| x),
)
}
#[cfg(all(test, feature = "parallel"))]
#[expect(clippy::unwrap_used, reason = "test code")]
fn build_and_recover(
items: &[crate::InternalValue],
parallel: Option<crate::table::writer::ParallelCompression>,
config: impl Fn(Writer) -> Writer,
open: impl Fn(&mut RecoverParams),
) -> crate::Result<(Table, tempfile::TempDir)> {
let dir = tempfile::tempdir()?;
let path = dir.path().join("table");
let mut writer = config(Writer::new(path.clone(), 0, 0, Arc::new(StdFs))?);
if let Some(parallel) = parallel {
writer = writer.use_parallel_compression(parallel);
}
for item in items {
writer.write(item.clone())?;
}
let (_, checksum) = writer.finish()?.unwrap();
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let table = {
let mut params = test_recover_params(path, checksum);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
open(&mut params);
Table::recover(params)?
};
Ok((table, dir))
}
#[cfg(all(test, feature = "parallel"))]
fn data_block_frames(table: &Table) -> crate::Result<Vec<Vec<u8>>> {
let file = std::fs::read(&*table.path)?;
table
.block_index
.iter()
.map(|keyed| {
let handle = keyed?;
let start = usize::try_from(*handle.offset()).expect("offset fits usize");
let end = start + handle.size() as usize;
Ok(file
.get(start..end)
.expect("the frame lies within the file")
.to_vec())
})
.collect()
}
#[cfg(feature = "parallel")]
#[test]
fn parallel_compression_matches_serial_output() -> crate::Result<()> {
use crate::table::writer::ParallelCompression;
let items: Vec<_> = (0u32..4000)
.map(|i| {
crate::InternalValue::from_components(
format!("key{i:08}").as_bytes(),
format!("value-{i}-some-payload-bytes").as_bytes(),
u64::from(i),
crate::ValueType::Value,
)
})
.collect();
let workers = |inline_below: Option<u32>| -> crate::Result<ParallelCompression> {
Ok(ParallelCompression {
spawner: Arc::new(crate::table::writer::RayonSpawner::with_threads(4)?),
threads: 4,
inline_below,
})
};
let check = |config: &dyn Fn(Writer) -> Writer,
open: &dyn Fn(&mut RecoverParams),
label: &str,
frames_vary: bool|
-> crate::Result<()> {
let (serial, _ds) = build_and_recover(&items, None, config, open)?;
let serial_rows: Vec<_> = serial.iter().collect::<crate::Result<_>>()?;
assert_eq!(
serial_rows.len(),
items.len(),
"{label}: all items must scan back"
);
let serial_frames = data_block_frames(&serial)?;
for inline_below in [Some(0), Some(u32::MAX), None] {
let (parallel, _dp) =
build_and_recover(&items, Some(workers(inline_below)?), config, open)?;
let variant = format!("{label}, inline below {inline_below:?}");
assert_eq!(
serial.metadata.data_block_count, parallel.metadata.data_block_count,
"{variant}: data_block_count must match"
);
let rows: Vec<_> = parallel.iter().collect::<crate::Result<_>>()?;
assert_eq!(
serial_rows, rows,
"{variant}: scan content/order must match"
);
if !frames_vary {
assert!(
serial_frames == data_block_frames(¶llel)?,
"{variant}: data block frames must be byte-identical",
);
}
for i in (0..items.len()).step_by(137) {
let key = format!("key{i:08}");
let hash = hash64(key.as_bytes());
assert_eq!(
serial.get(key.as_bytes(), crate::SeqNo::MAX, hash)?,
parallel.get(key.as_bytes(), crate::SeqNo::MAX, hash)?,
"{variant}: point read for {key} must match"
);
}
}
Ok(())
};
let as_written = |_: &mut RecoverParams| {};
check(&|w| w.use_data_block_size(256), &as_written, "plain", false)?;
check(
&|w| {
w.use_data_block_size(256)
.use_partitioned_filter()
.use_filter_partition_size(256)
},
&as_written,
"partitioned filter",
false,
)?;
check(
&|w| w.use_data_block_size(256).use_seqno_in_index(true),
&as_written,
"seqno_in_index",
false,
)?;
check(
&|w| {
w.use_data_block_size(256).use_kv_checksums(
crate::runtime_config::KvChecksumPolicy::AllLevels,
crate::runtime_config::ChecksumAlgorithm::Xxh3_64,
)
},
&as_written,
"per-KV checksums",
false,
)?;
check(
&|w| {
w.use_data_block_size(256)
.use_columnar(true)
.use_row_group_size(1_024)
},
&as_written,
"columnar",
true,
)?;
#[cfg(feature = "lz4")]
check(
&|w| {
w.use_data_block_size(256)
.use_data_block_compression(CompressionType::Lz4)
},
&as_written,
"lz4",
false,
)?;
#[cfg(feature = "zstd")]
{
let dictionary = Arc::new(make_test_dictionary());
let dict_id = dictionary.id();
check(
&|w| {
w.use_data_block_size(256)
.use_data_block_compression(CompressionType::ZstdDict { level: 3, dict_id })
.use_zstd_dictionary(Some(dictionary.clone()))
},
&|params| {
params.zstd_dictionaries =
crate::compression::ZstdDictionaries::new().with(dictionary.clone());
},
"zstd with a dictionary",
false,
)?;
}
#[cfg(feature = "page_ecc")]
check(
&|w| {
w.use_data_block_size(256).use_page_ecc(
true,
crate::runtime_config::EccScheme::Xor { data_shards: 4 },
)
},
&as_written,
"page ECC",
false,
)?;
#[cfg(feature = "encryption")]
{
let key: Arc<dyn crate::encryption::EncryptionProvider> =
Arc::new(crate::encryption::Aes256GcmProvider::new(&[7; 32]));
check(
&|w| w.use_data_block_size(256).use_encryption(Some(key.clone())),
&|params| params.encryption = Some(key.clone()),
"encryption",
true,
)?;
}
Ok(())
}
#[test]
fn zone_map_section_roundtrips_one_entry_per_block() -> crate::Result<()> {
let items: Vec<crate::InternalValue> = (0..200u32)
.map(|i| {
crate::InternalValue::from_components(
format!("k{i:05}").into_bytes(),
format!("v{i}").into_bytes(),
0,
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
let zm = &table.zone_map;
assert!(!zm.is_empty(), "zone map should be populated when enabled");
assert!(
zm.len() >= 2,
"rotation should yield several blocks, got {}",
zm.len()
);
Ok(())
},
Some(20),
Some(|w: Writer| w.use_zone_map(true)),
)
}
#[test]
fn zone_map_corrupt_section_falls_back_instead_of_failing_open() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = {
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_zone_map(true);
for i in 0..200u32 {
if i % 20 == 0 {
writer.spill_block()?;
}
writer.write(crate::InternalValue::from_components(
format!("k{i:05}").into_bytes(),
b"v".to_vec(),
0,
crate::ValueType::Value,
))?;
}
writer.finish()?.expect("table written").1
};
let recover = || -> crate::Result<Table> {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
#[cfg_attr(not(feature = "metrics"), expect(unused_mut))]
let mut params = test_recover_params(file.clone(), checksum);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)
};
let zm_handle = {
let table = recover()?;
assert!(!table.zone_map.is_empty(), "zone map should be populated");
table.regions.zone_map.expect("zone-map section present")
};
let mut bytes = std::fs::read(&file)?;
let corrupt_at = usize::try_from(zm_handle.offset().0).expect("offset fits usize") + 4;
*bytes
.get_mut(corrupt_at)
.expect("corruption offset within file") ^= 0xFF;
std::fs::write(&file, &bytes)?;
let table = recover()?;
assert!(
table.zone_map.is_empty(),
"corrupt zone-map section should disable block-skip, not fail open"
);
assert_eq!(
table.metadata.item_count, 200,
"the rest of the table must still load with a corrupt zone map"
);
Ok(())
}
#[test]
fn zone_map_absent_without_policy() -> crate::Result<()> {
let items: Vec<crate::InternalValue> = (0..50u32)
.map(|i| {
crate::InternalValue::from_components(
format!("k{i:05}").into_bytes(),
b"v".to_vec(),
0,
crate::ValueType::Value,
)
})
.collect();
test_with_table(
&items,
|table| {
assert!(
table.zone_map.is_empty(),
"no zone map should be loaded without the policy"
);
Ok(())
},
None,
None::<fn(Writer) -> Writer>,
)
}
fn recover_test_table(file: &std::path::Path, checksum: Checksum) -> crate::Result<Table> {
recover_test_table_with_id(file, checksum, 0)
}
fn recover_test_table_with_id(
file: &std::path::Path,
checksum: Checksum,
table_id: TableId,
) -> crate::Result<Table> {
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(file.to_path_buf(), checksum);
params.table_id = table_id;
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)
}
#[test]
fn dropping_a_deleted_table_removes_its_restrict_bound_sidecar() -> crate::Result<()> {
use crate::fs::Fs;
let dir = tempdir()?;
let file = dir.path().join("0");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(crate::InternalValue::from_components(
b"k",
b"v",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer.finish()?.expect("table written");
let fs = StdFs;
crate::restrict_bound::write(&fs, &file, None, 0, b"k", crate::fs::SyncMode::Normal)?;
let sidecar = crate::restrict_bound::sidecar_path(&file);
assert!(fs.exists(&sidecar)?, "sidecar present before retirement");
let table = recover_test_table(&file, checksum)?;
table.mark_as_deleted();
drop(table);
assert!(
!fs.exists(&sidecar)?,
"retiring the table must reclaim its restrict-bound sidecar, not leak it",
);
Ok(())
}
#[test]
fn delete_bitmap_section_round_trips() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let keys: [&[u8]; 8] = [b"a", b"b", b"c", b"d", b"e", b"f", b"g", b"h"];
let deleted_rows = [0u32, 2, 5];
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_zone_map(true);
for key in keys {
writer.write(crate::InternalValue::from_components(
key,
b"v",
1,
crate::ValueType::Value,
))?;
}
for &row in &deleted_rows {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert!(
table.regions.delete_bitmap.is_some(),
"delete-bitmap section must be present when rows are deleted"
);
let dv = table.delete_bitmap();
assert_eq!(dv.len(), deleted_rows.len() as u64);
for row in 0..8u32 {
assert_eq!(
dv.contains(row),
deleted_rows.contains(&row),
"row {row} membership mismatch after reopen"
);
}
Ok(())
}
#[test]
fn delete_bitmap_section_absent_when_no_deletes() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?;
writer.write(crate::InternalValue::from_components(
b"a",
b"v",
1,
crate::ValueType::Value,
))?;
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert!(
table.regions.delete_bitmap.is_none(),
"no delete-bitmap section when the segment has no deletes"
);
assert!(table.delete_bitmap().is_empty());
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn columnar_zone_map_records_per_column_stats_and_round_trips() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
for i in 0..32u32 {
let key = format!("k{i:04}").into_bytes();
let value = format!("v{i:04}").into_bytes();
writer.write(crate::InternalValue::from_components(
key,
value,
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert!(table.metadata.columnar, "written as a columnar table");
assert!(!table.zone_map.is_empty(), "zone map populated");
let mut saw_value_column = false;
for handle in table.block_index.iter() {
let handle = handle?;
let stats = table
.zone_map
.columns_for(handle.offset().0)
.expect("every data block has a zone-map entry");
let ids: Vec<u32> = stats.iter().map(|s| s.column_id).collect();
assert!(
ids.contains(&u32::from(COL_USER_KEY)),
"the user-key column is recorded, got ids {ids:?}",
);
if let Some(v) = stats.iter().find(|s| s.column_id == u32::from(COL_VALUE)) {
saw_value_column = true;
assert!(v.min <= v.max, "the value column's range is ordered");
assert!(!v.min.is_empty(), "the value column carries a real range");
}
}
assert!(
saw_value_column,
"at least one block records the non-key value column's stats",
);
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("the honest columnar table must pass every gate, {gate:?} refused it: {e}");
}
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_rows_in_columnar_scan() -> crate::Result<()> {
use crate::table::columnar::{
COL_SEQNO, COL_USER_KEY, COL_VALUE, COL_VALUE_TYPE, column_batch_to_entries,
};
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 64u32;
let deleted = [3u32, 10, 50];
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
for i in 0..n {
let key = format!("k{i:04}").into_bytes();
writer.write(crate::InternalValue::from_components(
key,
b"v",
1,
crate::ValueType::Value,
))?;
}
for &row in &deleted {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let batches = table
.columnar_scan(&[COL_USER_KEY, COL_SEQNO, COL_VALUE_TYPE, COL_VALUE], None)?
.collect::<crate::Result<Vec<_>>>()?;
let mut got: Vec<Vec<u8>> = Vec::new();
for batch in &batches {
for entry in column_batch_to_entries(batch)? {
got.push(entry.key.user_key.to_vec());
}
}
let expected: Vec<Vec<u8>> = (0..n)
.filter(|i| !deleted.contains(i))
.map(|i| format!("k{i:04}").into_bytes())
.collect();
assert_eq!(
got, expected,
"deleted row positions must be masked out of the columnar scan"
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
#[expect(clippy::unwrap_used)]
fn restricted_columnar_scan_skips_punched_prefix_and_masks_sub_bound_rows() -> crate::Result<()> {
use crate::table::columnar::{
COL_SEQNO, COL_USER_KEY, COL_VALUE, COL_VALUE_TYPE, column_batch_to_entries,
};
use std::io::{Seek, SeekFrom, Write as _};
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 256u32;
let keys: Vec<Vec<u8>> = (0..n).map(|i| format!("k{i:04}").into_bytes()).collect();
let deleted_punched = 1u32;
let deleted_live = 250u32;
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(256);
for key in &keys {
writer.write(crate::InternalValue::from_components(
key.as_slice(),
b"v",
1,
crate::ValueType::Value,
))?;
}
writer.delete_bitmap_mut().insert(deleted_punched);
writer.delete_bitmap_mut().insert(deleted_live);
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let handles: Vec<_> = table
.block_index
.iter()
.collect::<crate::Result<Vec<_>>>()?;
assert!(handles.len() >= 3, "need several blocks to punch a prefix");
let first_end = handles[0].end_key().to_vec();
let j = keys.iter().position(|k| *k == first_end).unwrap();
let bound_idx = j + 2;
assert!(
bound_idx < deleted_live as usize,
"the live deleted row must stay above the bound"
);
let bound = crate::UserKey::from(keys[bound_idx].as_slice());
let punch_off = table.punch_offset_for(&bound)?;
assert_eq!(
punch_off,
handles[1].offset().0,
"exactly the first block is below the bound"
);
let restricted = table.with_restriction(bound);
let mut f = std::fs::OpenOptions::new().write(true).open(&file)?;
f.seek(SeekFrom::Start(0))?;
f.write_all(&vec![0u8; usize::try_from(punch_off).unwrap()])?;
f.sync_all()?;
let batches = restricted
.columnar_scan(&[COL_USER_KEY, COL_SEQNO, COL_VALUE_TYPE, COL_VALUE], None)?
.collect::<crate::Result<Vec<_>>>()?;
let mut got: Vec<Vec<u8>> = Vec::new();
for batch in &batches {
for entry in column_batch_to_entries(batch)? {
got.push(entry.key.user_key.to_vec());
}
}
let expected: Vec<Vec<u8>> = (bound_idx..n as usize)
.filter(|&i| i != deleted_live as usize)
.map(|i| keys[i].clone())
.collect();
assert_eq!(
got, expected,
"scan must start at the bound, mask the straddling block's sub-bound \
rows, and keep delete positions aligned across the punched prefix"
);
Ok(())
}
#[cfg(feature = "columnar")]
fn cursor_keys(
cursor: crate::table::columnar_cursor::ColumnarCursor,
) -> crate::Result<Vec<Vec<u8>>> {
use crate::table::columnar::{COL_USER_KEY, bytes_column_row};
let mut keys = Vec::new();
for batch in cursor {
let batch = batch?;
let column = batch
.columns
.iter()
.find(|c| c.column_id == COL_USER_KEY)
.expect("the key column is projected");
for row in 0..batch.row_count {
keys.push(bytes_column_row(&column.data, batch.row_count, row)?.to_vec());
}
}
Ok(keys)
}
#[cfg(feature = "columnar")]
#[test]
fn a_key_range_over_positional_deletes_keeps_their_positions_past_the_skipped_groups()
-> crate::Result<()> {
use crate::table::columnar::COL_USER_KEY;
use core::ops::Bound;
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 256u32;
let key = |i: u32| format!("k{i:04}").into_bytes();
let deleted = [3u32, 200];
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(256);
for i in 0..n {
writer.write(crate::InternalValue::from_components(
key(i),
b"v",
1,
crate::ValueType::Value,
))?;
}
for &row in &deleted {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let cursor = table.columnar_cursor(
&[COL_USER_KEY],
None,
Bound::Included(crate::UserKey::from(key(100))),
Bound::Unbounded,
None,
)?;
let got: Vec<Vec<u8>> = cursor_keys(cursor)?
.into_iter()
.filter(|k| k.as_slice() >= key(100).as_slice())
.collect();
let expected: Vec<Vec<u8>> = (100..n).filter(|&i| i != 200).map(key).collect();
assert_eq!(got, expected);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_straddling_groups_key_pages_count_against_the_share() -> crate::Result<()> {
use crate::table::columnar::COL_VALUE;
use core::ops::Bound;
let dir = tempdir()?;
let file = dir.path().join("table");
let key = |i: u32| format!("{i:0200}").into_bytes();
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
for i in 0..64 {
writer.write(crate::InternalValue::from_components(
key(i),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let restricted = table.with_restriction(crate::UserKey::from(key(5)));
const SHARE: u64 = 2_048;
let mut cursor = restricted.columnar_cursor(
&[COL_VALUE],
None,
Bound::Unbounded,
Bound::Unbounded,
Some(SHARE),
)?;
let mut rows = 0;
for batch in &mut cursor {
rows += batch?.row_count;
}
assert_eq!(rows, 59, "the rows at or past the bound");
assert!(
cursor.take_oversized() > 0,
"the key pages read beside the values are past the {SHARE} B share",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_restricted_table_read_in_runs_masks_the_straddling_rows_of_every_run() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use core::ops::Bound;
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 256u32;
let key = |i: u32| format!("k{i:04}").into_bytes();
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(4_096)
.use_columnar_page_size(256);
for i in 0..n {
writer.write(crate::InternalValue::from_components(
key(i),
b"value",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let bound = 40u32;
let restricted = table.with_restriction(crate::UserKey::from(key(bound)));
let cursor = restricted.columnar_cursor(
&[COL_USER_KEY, COL_VALUE],
None,
Bound::Unbounded,
Bound::Unbounded,
Some(1),
)?;
let expected: Vec<Vec<u8>> = (bound..n).map(key).collect();
assert_eq!(cursor_keys(cursor)?, expected);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
#[expect(clippy::expect_used, reason = "test code")]
fn unshare_for_heal_preserves_unpunched_blocks_of_a_restricted_table() -> crate::Result<()> {
use crate::fs::{Fs, MemFs};
use crate::table::block_index::BlockIndex;
use std::sync::Arc;
let memfs = Arc::new(MemFs::new());
let fs: Arc<dyn Fs> = memfs.clone();
let root = std::path::absolute("/db")?;
memfs.create_dir_all(&root)?;
let build = |name: &str| -> crate::Result<Table> {
let path = root.join(name);
let mut writer = Writer::new(path.clone(), 0, 0, Arc::clone(&fs))?.use_data_block_size(256);
for i in 0..256u32 {
writer.write(crate::InternalValue::from_components(
format!("k{i:04}").into_bytes(),
b"v",
1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
#[cfg(feature = "metrics")]
let metrics = Arc::new(Metrics::default());
let mut params = test_recover_params(path, checksum);
params.descriptor_table = None;
params.fs = Arc::clone(&fs);
#[cfg(feature = "metrics")]
{
params.metrics = metrics;
}
Table::recover(params)
};
let all_zero = |path: &std::path::Path, off: u64, len: usize| -> crate::Result<bool> {
let file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
let bytes = crate::file::read_exact(&*file, off, len)?;
Ok(bytes.iter().all(|&b| b == 0))
};
let table = build("0")?;
let handles: Vec<_> = table
.block_index
.iter()
.collect::<crate::Result<Vec<_>>>()?;
assert!(handles.len() >= 3, "need several blocks to restrict over");
let bound = handles.get(1).expect("second block").end_key().clone();
let (b0_off, b0_len) = {
let h = handles.first().expect("first block");
(h.offset().0, h.size() as usize)
};
let restricted = table.with_restriction(bound.clone());
let source = fs.open(
&restricted.path,
&crate::fs::FsOpenOptions::new().read(true),
)?;
restricted
.unshare_for_heal(&*source, crate::fs::SyncMode::Normal)
.expect("unshare succeeds");
assert!(
!all_zero(&restricted.path, b0_off, b0_len)?,
"an unpunched restricted table's prefix blocks must be copied verbatim, \
not turned into holes the (missing) sidecar does not cover"
);
let table = build("1")?;
let handles: Vec<_> = table
.block_index
.iter()
.collect::<crate::Result<Vec<_>>>()?;
let punch = table.punch_offset_for(&bound)?;
for h in &handles {
if h.offset().0 < punch {
memfs.punch_hole(&table.path, h.offset().0, u64::from(h.size()))?;
}
}
let (p0_off, p0_len) = {
let h = handles.first().expect("first block");
(h.offset().0, h.size() as usize)
};
let restricted = table.with_restriction(bound);
let source = fs.open(
&restricted.path,
&crate::fs::FsOpenOptions::new().read(true),
)?;
restricted
.unshare_for_heal(&*source, crate::fs::SyncMode::Normal)
.expect("unshare succeeds");
assert!(
all_zero(&restricted.path, p0_off, p0_len)?,
"a genuinely punched extent stays a hole in the heal copy"
);
let table = build("2")?;
let handles: Vec<_> = table
.block_index
.iter()
.collect::<crate::Result<Vec<_>>>()?;
assert!(handles.len() >= 5, "need blocks on both sides of the gap");
let bound = handles.get(3).expect("fourth block").end_key().clone();
let extent = |i: usize| {
let h = handles.get(i).expect("block");
(h.offset().0, h.size() as usize)
};
let (hole_a, kept, hole_b) = (extent(0), extent(1), extent(2));
for (off, len) in [hole_a, hole_b] {
memfs.punch_hole(&table.path, off, len as u64)?;
}
let restricted = table.with_restriction(bound);
let source = fs.open(
&restricted.path,
&crate::fs::FsOpenOptions::new().read(true),
)?;
restricted
.unshare_for_heal(&*source, crate::fs::SyncMode::Normal)
.expect("unshare succeeds");
assert!(
all_zero(&restricted.path, hole_a.0, hole_a.1)?
&& all_zero(&restricted.path, hole_b.0, hole_b.1)?,
"both punched extents stay holes in the heal copy"
);
assert!(
!all_zero(&restricted.path, kept.0, kept.1)?,
"the intact block between two holes is copied, not merged into a hole"
);
Ok(())
}
#[test]
fn copy_on_write_strategy_suppresses_the_delete_bitmap_section() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.delete_strategy(crate::config::DeleteStrategy::CopyOnWrite);
for key in [b"a".as_ref(), b"b", b"c"] {
writer.write(crate::InternalValue::from_components(
key,
b"v",
1,
crate::ValueType::Value,
))?;
}
writer.delete_bitmap_mut().insert(1);
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert!(
table.regions.delete_bitmap.is_none(),
"copy-on-write must not persist a delete-bitmap section"
);
assert!(table.delete_bitmap().is_empty());
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_rows_in_range_scan() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 64u32;
let deleted = [3u32, 10, 50];
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
for i in 0..n {
let key = format!("k{i:04}").into_bytes();
writer.write(crate::InternalValue::from_components(
key,
b"v",
1,
crate::ValueType::Value,
))?;
}
for &row in &deleted {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
let got: Vec<Vec<u8>> = table
.range_iter(..)
.map(|r| r.map(|kv| kv.key.user_key.to_vec()))
.collect::<crate::Result<Vec<_>>>()?;
let expected: Vec<Vec<u8>> = (0..n)
.filter(|i| !deleted.contains(i))
.map(|i| format!("k{i:04}").into_bytes())
.collect();
assert_eq!(
got, expected,
"deleted row positions must be masked out of the range scan"
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_deleted_key_in_point_read() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let n = 64u32;
let deleted = [3u32, 10, 50];
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
for i in 0..n {
let key = format!("k{i:04}").into_bytes();
writer.write(crate::InternalValue::from_components(
key,
b"v",
1,
crate::ValueType::Value,
))?;
}
for &row in &deleted {
writer.delete_bitmap_mut().insert(row);
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
for i in 0..n {
let key = format!("k{i:04}").into_bytes();
let got = table.get(&key, SeqNo::MAX, hash64(&key))?;
if deleted.contains(&i) {
assert!(got.is_none(), "deleted key {i} must read as absent");
} else {
assert!(got.is_some(), "live key {i} must be found");
}
}
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_stores_value_subcolumns_and_round_trips() -> crate::Result<()> {
use crate::table::columnar::{Column, TypeTag, entries_to_column_batch, unframe_value_cells};
let dir = tempdir()?;
let file = dir.path().join("table");
let mut batch = entries_to_column_batch(&[
crate::InternalValue::from_components(b"k0", b"ignored", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"k1", b"ignored", 0, crate::ValueType::Value),
])?;
batch.columns.pop();
batch.columns.push(Column {
column_id: 3,
type_tag: TypeTag::Fixed(4),
validity: None,
data: vec![1, 0, 0, 0, 2, 0, 0, 0].into(),
});
let mut bytes_data = Vec::new();
for off in [0u32, 2, 5] {
bytes_data.extend_from_slice(&off.to_le_bytes());
}
bytes_data.extend_from_slice(b"aabbb");
batch.columns.push(Column {
column_id: 4,
type_tag: TypeTag::Bytes,
validity: None,
data: bytes_data.into(),
});
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert!(table.metadata.columnar, "segment must be columnar");
let tags = [TypeTag::Fixed(4), TypeTag::Bytes];
let v0 = table
.get(b"k0", SeqNo::MAX, hash64(b"k0"))?
.expect("k0 present");
assert_eq!(
unframe_value_cells(v0.value.as_ref(), &tags)?,
vec![&[1, 0, 0, 0][..], &b"aa"[..]],
);
let v1 = table
.get(b"k1", SeqNo::MAX, hash64(b"k1"))?
.expect("k1 present");
assert_eq!(
unframe_value_cells(v1.value.as_ref(), &tags)?,
vec![&[2, 0, 0, 0][..], &b"bbb"[..]],
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn delete_bitmap_masks_value_subcolumns_in_point_and_projection_reads() -> crate::Result<()> {
use crate::fs::SyncMode;
use crate::table::columnar::{Column, TypeTag, entries_to_column_batch, unframe_value_cells};
use crate::table::delete_bitmap::DeleteBitmap;
let dir = tempdir()?;
let src = dir.path().join("src");
let out = dir.path().join("out");
let fixed: [u32; 5] = [10, 20, 30, 40, 50];
let payloads: [&[u8]; 5] = [b"a", b"bb", b"ccc", b"dddd", b"eeeee"];
let mut batch = entries_to_column_batch(
&(0..5u32)
.map(|i| {
crate::InternalValue::from_components(
format!("k{i}").into_bytes(),
b"x",
0, crate::ValueType::Value,
)
})
.collect::<Vec<_>>(),
)?;
batch.columns.pop();
batch.columns.push(Column {
column_id: 3,
type_tag: TypeTag::Fixed(4),
validity: None,
data: fixed.iter().flat_map(|v| v.to_le_bytes()).collect(),
});
let mut bytes_data = Vec::new();
let mut acc = 0u32;
bytes_data.extend_from_slice(&acc.to_le_bytes());
for p in payloads {
acc += u32::try_from(p.len()).unwrap();
bytes_data.extend_from_slice(&acc.to_le_bytes());
}
for p in payloads {
bytes_data.extend_from_slice(p);
}
batch.columns.push(Column {
column_id: 4,
type_tag: TypeTag::Bytes,
validity: None,
data: bytes_data.into(),
});
let mut writer = Writer::new(src.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
let (_, checksum) = writer.finish()?.expect("source written");
let source = recover_test_table(&src, checksum)?;
let mut bitmap = DeleteBitmap::new();
bitmap.insert(1);
bitmap.insert(3);
let out_checksum = source
.relocate_columnar_with_deletes(
|| {
use crate::fs::{Fs, FsOpenOptions};
Ok(crate::table::file_pool::ClaimedFile {
id: 1,
path: out.clone(),
file: StdFs.open(&out, &FsOpenOptions::new().write(true).create_new(true))?,
durable_entry: false,
writing: crate::table::file_pool::Writing::alone(1),
})
},
&StdFs,
&bitmap,
SyncMode::Normal,
None,
)?
.checksum;
let relocated = recover_test_table_with_id(&out, out_checksum, 1)?;
let tags = [TypeTag::Fixed(4), TypeTag::Bytes];
for i in 0..5u32 {
let key = format!("k{i}").into_bytes();
let got = relocated.get(&key, SeqNo::MAX, hash64(&key))?;
if i == 1 || i == 3 {
assert!(got.is_none(), "masked row {i} must read absent");
} else {
let v = got.expect("survivor present");
assert_eq!(
unframe_value_cells(v.value.as_ref(), &tags)?,
vec![&fixed[i as usize].to_le_bytes()[..], payloads[i as usize]],
"survivor {i} sub-cells",
);
}
}
let batches = relocated
.columnar_scan(&[3], None)?
.collect::<crate::Result<Vec<_>>>()?;
let mut col3 = Vec::new();
let mut rows = 0u32;
for b in &batches {
assert!(
b.columns.iter().all(|c| c.column_id == 3),
"projection decodes only sub-column 3",
);
rows += b.row_count;
for c in b.columns.iter().filter(|c| c.column_id == 3) {
col3.extend_from_slice(&c.data);
}
}
assert_eq!(rows, 3, "two of five rows masked out of the projection");
let want: Vec<u8> = [10u32, 30, 50]
.iter()
.flat_map(|v| v.to_le_bytes())
.collect();
assert_eq!(col3, want, "projected fixed bytes are the survivors");
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_accounts_tombstones_seqno_bounds_and_restart_locator() -> crate::Result<()>
{
use crate::config::{LocatorPolicyEntry, LocatorPrecision};
use crate::table::columnar::{Column, TypeTag, entries_to_column_batch};
let dir = tempdir()?;
let file = dir.path().join("t");
let mut batch = entries_to_column_batch(&[
crate::InternalValue::from_components(b"k0", b"v", 0, crate::ValueType::Value),
crate::InternalValue::from_components(b"k1", b"", 0, crate::ValueType::Tombstone),
crate::InternalValue::from_components(b"k2", b"", 0, crate::ValueType::WeakTombstone),
])?;
batch.columns.pop();
batch.columns.push(Column {
column_id: 3,
type_tag: TypeTag::Fixed(2),
validity: None,
data: vec![1, 1, 2, 2, 3, 3].into(),
});
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_seqno_in_index(true)
.use_locator(LocatorPolicyEntry::Enabled {
precision: LocatorPrecision::Restart,
block_id_bits: None,
slot_bits: None,
});
writer.write_columnar_batch(&batch, &crate::comparator::default_comparator())?;
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
assert_eq!(table.metadata.tombstone_count, 2, "two tombstone-kind rows");
assert_eq!(table.metadata.weak_tombstone_count, 1, "one weak tombstone");
assert_eq!(
table.metadata.seqnos,
(0, 0),
"columnar ingest writes local seqno bounds of (0, 0)",
);
assert!(
table.get(b"k0", SeqNo::MAX, hash64(b"k0"))?.is_some(),
"the live row reads back",
);
Ok(())
}
#[test]
fn reconcile_gates_lowered_tli_separator_rejects_with_separators_gate() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("t");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?.use_data_block_size(128);
for i in 0u64..64 {
writer.write(crate::InternalValue::from_components(
alloc::format!("key-{i:04}").into_bytes(),
alloc::format!("v{i:04}").into_bytes(),
i + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("intact separators must pass every gate, {gate:?} refused it: {e}");
}
crate::test_forge::forge_tli_mirrors_lower_first_separator(&file, 0, None)?;
let table = recover_test_table(&file, checksum)?;
let result = table.verify_reconcile_gates(None, false);
assert!(
matches!(
result,
Err((
crate::table::ReconcileGate::Separators,
crate::Error::InvalidHeader(
"tli separator does not match the addressed block's decoded last key"
)
))
),
"a lowered separator must be rejected by the separator gate, got {:?}",
result.map_err(|(gate, e)| (gate, e.to_string())),
);
Ok(())
}
#[test]
fn reconcile_gates_stale_kv_footer_rejects_with_kv_checksums_gate() -> crate::Result<()> {
use crate::runtime_config::{ChecksumAlgorithm, KvChecksumPolicy};
let dir = tempdir()?;
let file = dir.path().join("t");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_kv_checksums(KvChecksumPolicy::AllLevels, ChecksumAlgorithm::Xxh3_64);
for i in 0u64..40 {
writer.write(crate::InternalValue::from_components(
alloc::format!("key-{i:04}").into_bytes(),
alloc::format!("v{i:04}").into_bytes(),
i + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("an intact footer must pass every gate, {gate:?} refused it: {e}");
}
crate::test_forge::forge_stale_kv_footer(&file)?;
let table = recover_test_table(&file, checksum)?;
let result = table.verify_reconcile_gates(None, false);
assert!(
matches!(
result,
Err((
crate::table::ReconcileGate::KvChecksums,
crate::Error::ChecksumMismatch { .. }
))
),
"a stale per-KV footer must be rejected by the per-KV gate, got {:?}",
result.map_err(|(gate, e)| (gate, e.to_string())),
);
Ok(())
}
#[test]
fn verify_locator_rejects_a_redirected_key_mapping() -> crate::Result<()> {
use crate::config::{LocatorPolicyEntry, LocatorPrecision};
use crate::table::locator::{LocatorSpec, build_locator_section};
let dir = tempdir()?;
let file = dir.path().join("t");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_data_block_size(128)
.use_locator(LocatorPolicyEntry::Enabled {
precision: LocatorPrecision::Block,
block_id_bits: None,
slot_bits: None,
});
for i in 0u64..200 {
writer.write(crate::InternalValue::from_components(
format!("key-{i:04}").into_bytes(),
format!("v{i:04}").into_bytes(),
i + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("table written");
let table = recover_test_table(&file, checksum)?;
if let Err((gate, e)) = table.verify_reconcile_gates(None, false) {
panic!("an intact locator must pass every gate, {gate:?} refused it: {e}");
}
let mut triples: Vec<(u64, u64, u64)> = Vec::new();
let block_count = table.block_index.iter().count() as u64;
assert!(block_count >= 2, "need multiple blocks to redirect between");
let mut seen = crate::HashSet::default();
for (ordinal, handle) in table.block_index.iter().enumerate() {
let handle = handle?;
let block_handle = crate::table::BlockHandle::new(handle.offset(), handle.size());
let entries = table.decode_block_entries(&block_handle)?;
for e in entries {
let uk = e.key.user_key.to_vec();
if seen.insert(uk.clone()) {
triples.push((crate::hash::hash64(&uk), ordinal as u64, 0));
}
}
}
let orig = triples[0].1;
triples[0].1 = if orig == 0 { block_count - 1 } else { 0 };
let spec = LocatorSpec {
precision: LocatorPrecision::Block,
block_id_bits: None,
slot_bits: None,
};
let forged = build_locator_section(&triples, spec).expect("forged section builds");
crate::test_forge::forge_replace_section_payload(&file, b"locator", &forged, None)?;
let table = recover_test_table(&file, checksum)?;
let result = table.verify_reconcile_gates(None, false);
assert!(
matches!(
result,
Err((
crate::table::ReconcileGate::Locator,
crate::Error::InvalidHeader(_)
))
),
"a redirected locator must be rejected by the locator gate, got {:?}",
result.map_err(|(gate, e)| (gate, e.to_string())),
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_on_an_empty_batch_writes_no_block() -> crate::Result<()> {
use crate::table::columnar::{Column, TypeTag};
let dir = tempdir()?;
let file = dir.path().join("t");
let empty = crate::table::columnar::ColumnBatch {
row_count: 0,
columns: vec![
Column {
column_id: 0,
type_tag: TypeTag::Bytes,
validity: None,
data: 0u32.to_le_bytes().to_vec().into(),
},
Column {
column_id: 1,
type_tag: TypeTag::Number(crate::table::columnar::Number::U64_LE),
validity: None,
data: Vec::new().into(),
},
Column {
column_id: 2,
type_tag: TypeTag::Fixed(1),
validity: None,
data: Vec::new().into(),
},
Column {
column_id: 3,
type_tag: TypeTag::Fixed(4),
validity: None,
data: Vec::new().into(),
},
],
};
let mut writer = Writer::new(file, 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true);
assert!(
writer
.write_columnar_batch(&empty, &crate::comparator::default_comparator())?
.is_none(),
"an empty batch yields no last key",
);
assert!(
writer.finish()?.is_none(),
"an empty batch must not produce an SST",
);
Ok(())
}
#[test]
fn recover_salvage_propagates_a_transient_locator_read() -> crate::Result<()> {
use crate::config::{LocatorPolicyEntry, LocatorPrecision};
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
use crate::io::ErrorKind;
let dir = tempdir()?;
let sst = dir.path().join("0");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
{
let mut w = Writer::new(sst.clone(), 0, 0, Arc::clone(&fs))?
.use_data_block_size(128)
.use_locator(LocatorPolicyEntry::Enabled {
precision: LocatorPrecision::Entry,
block_id_bits: None,
slot_bits: None,
});
for i in 0..256u32 {
w.write(crate::InternalValue::from_components(
format!("k{i:05}").into_bytes(),
format!("v{i}").into_bytes(),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
assert!(w.finish()?.is_some(), "the SST is non-empty");
}
let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &sst)?);
let loc_offset = {
let live = {
let mut params = test_recover_params(sst.clone(), checksum);
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
live.regions
.locator
.expect("the multi-block SST carries a locator section")
.offset()
.0
};
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
injector.arm(
FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Interrupted))
.at_offset(loc_offset)
.once(),
);
let faulted: Arc<dyn crate::fs::Fs> = Arc::new(fault);
let result = {
let mut params = test_recover_params(sst, checksum);
params.fs = Arc::clone(&faulted);
Table::recover_inner(
params,
RecoveryMode::Salvage {
expected_id: None,
prefer_mid_meta: false,
},
)
};
injector.clear();
assert!(
matches!(&result, Err(crate::Error::Io(e)) if e.kind() == ErrorKind::Interrupted),
"a transient locator read in salvage mode must propagate (not degrade to a \
section-degraded open that later fails as FeatureUnsupported): {result:?}",
);
Ok(())
}
#[test]
fn recover_salvage_degrades_a_persistent_filter_index_read() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
use crate::io::ErrorKind;
let dir = tempdir()?;
let sst = dir.path().join("0");
let fs: Arc<dyn crate::fs::Fs> = Arc::new(StdFs);
{
let mut w = Writer::new(sst.clone(), 0, 0, Arc::clone(&fs))?
.use_data_block_size(128)
.use_partitioned_filter();
for i in 0..256u32 {
w.write(crate::InternalValue::from_components(
format!("k{i:05}").into_bytes(),
format!("v{i}").into_bytes(),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
assert!(w.finish()?.is_some(), "the SST is non-empty");
}
let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(&*fs, &sst)?);
let tli_offset = {
let live = {
let mut params = test_recover_params(sst.clone(), checksum);
params.fs = Arc::clone(&fs);
Table::recover(params)?
};
live.regions
.filter_tli
.expect("the partitioned-filter SST carries a filter_tli section")
.offset()
.0
};
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
injector.arm(
FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other))
.at_offset(tli_offset)
.once(),
);
let faulted: Arc<dyn crate::fs::Fs> = Arc::new(fault);
let result = {
let mut params = test_recover_params(sst, checksum);
params.fs = Arc::clone(&faulted);
Table::recover_inner(
params,
RecoveryMode::Salvage {
expected_id: None,
prefer_mid_meta: false,
},
)
};
injector.clear();
let recovered = match result {
Ok(table) => table,
Err(e) => panic!(
"a persistent filter-index read in salvage mode must degrade the rebuildable \
section and recover the table, not propagate and drop it: {e:?}",
),
};
assert!(
recovered.salvage_degraded_a_rebuildable_section(),
"the recovered table must report the degraded rebuildable section",
);
let key = b"k00010";
let found = recovered.get(key, SeqNo::MAX, crate::hash::hash64(key))?;
assert_eq!(
found.map(|item| item.value.to_vec()),
Some(b"v10".to_vec()),
"a key the table holds is found without the filter index",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_columnar_table_in_a_superseded_layout_is_refused_at_open() -> crate::Result<()> {
let dir = tempdir()?;
let sst = dir.path().join("table");
let mut writer = Writer::new(sst.clone(), 0, 0, Arc::new(StdFs))?.use_columnar(true);
for i in 0..64_u32 {
writer.write(InternalValue::from_components(
alloc::format!("key{i:05}").into_bytes(),
b"v".as_slice(),
0,
crate::ValueType::Value,
))?;
}
assert!(writer.finish()?.is_some(), "table is non-empty");
crate::test_forge::forge_meta_value_both_mirrors(&sst, b"descriptor#columnar_format", &[1])?;
let checksum = Checksum::from_raw(crate::repair::compute_table_checksum(&StdFs, &sst)?);
let Err(err) = Table::recover(test_recover_params(sst, checksum)) else {
panic!("a table in a superseded columnar layout must not open");
};
assert!(
matches!(
err,
crate::Error::UnsupportedFormat {
part: crate::FormatPart::Columnar,
found: Some(1),
expected: crate::table::meta::COLUMNAR_FORMAT_VERSION,
}
),
"expected the columnar format refusal, got {err:?}",
);
assert!(
err.is_environmental(),
"repair must propagate the refusal rather than grade the table as damaged",
);
Ok(())
}
#[cfg(feature = "columnar")]
fn columnar_table_file(
file: &std::path::Path,
rows: u32,
key_len: usize,
value_len: usize,
group_size: u32,
) -> crate::Result<Checksum> {
columnar_table_file_paged(
file,
rows,
key_len,
value_len,
group_size,
4 * 1_024 * 1_024,
)
}
#[cfg(feature = "columnar")]
fn columnar_table_file_paged(
file: &std::path::Path,
rows: u32,
key_len: usize,
value_len: usize,
group_size: u32,
page_size: u32,
) -> crate::Result<Checksum> {
let mut writer = Writer::new(file.to_path_buf(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_row_group_size(group_size)
.use_columnar_page_size(page_size);
for i in 0..rows {
let mut key = alloc::format!("key{i:06}").into_bytes();
key.resize(key_len, b'k');
writer.write(InternalValue::from_components(
key,
alloc::vec![b'v'; value_len],
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
Ok(checksum)
}
#[cfg(feature = "columnar")]
fn page_extent(file: &std::path::Path, group: &BlockHandle, column_id: u16) -> (usize, usize) {
use crate::coding::Decode;
use crate::table::block::Header;
let bytes = std::fs::read(file).expect("read table");
let at = usize::try_from(*group.offset()).expect("offset fits");
let frame = bytes.get(at..).expect("group within the file");
let header = Header::decode_from(&mut &frame[..]).expect("directory header");
let payload_at = Header::header_len(header.block_type);
let payload = frame
.get(payload_at..payload_at + header.data_length as usize)
.expect("directory payload");
let directory =
crate::table::column_page::PageDirectory::decode(payload).expect("decode directory");
let entry = directory
.entries()
.iter()
.find(|e| e.id.column_id == column_id)
.expect("the group has a page for the column");
let directory_len = header.on_disk_size_with(None) as usize;
(
at + directory_len + directory.pages_start() as usize + entry.offset as usize,
entry.length as usize,
)
}
#[cfg(feature = "columnar")]
#[test]
fn a_columnar_writer_cuts_at_the_row_group_size_and_a_row_writer_at_the_block_size()
-> crate::Result<()> {
let dir = tempdir()?;
let blocks = |columnar: bool| -> crate::Result<usize> {
let file = dir.path().join(if columnar { "columnar" } else { "rows" });
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(columnar)
.use_data_block_size(256)
.use_row_group_size(64 * 1_024);
for i in 0..400_u32 {
writer.write(InternalValue::from_components(
alloc::format!("key{i:06}").into_bytes(),
alloc::vec![b'v'; 64],
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer
.finish()?
.expect("finish() returns Some after writing data items");
let table = Table::recover(test_recover_params(file, checksum))?;
Ok(table.data_block_handles().count())
};
assert_eq!(blocks(true)?, 1, "the columnar table is one row group");
assert!(
blocks(false)? >= 50,
"the row-major table is cut at the data block size",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_projection_never_reads_a_corrupt_page_it_does_not_want() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file(&file, 200, 12, 256, 64 * 1_024)?;
let group = {
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let first = table
.data_block_handles()
.next()
.expect("the table has a row group")?;
*first.as_ref()
};
let (value_at, value_len) = page_extent(&file, &group, COL_VALUE);
let mut bytes = std::fs::read(&file)?;
let target = value_at + value_len / 2;
let byte = bytes.get_mut(target).expect("byte within the page");
*byte ^= 0x5A;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let keys = table.load_row_group(
&group,
&PageWant::projected(&[COL_USER_KEY], crate::table::row_group::RowPageSelect::All),
ReadCharge::Foreground,
)?;
assert!(keys.row_count() > 0, "the key projection reads its rows");
let table = Table::recover(test_recover_params(file, checksum))?;
assert!(
table
.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)
.is_err(),
"a read that wants the corrupt page must fail its check",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_projection_decodes_the_same_columns_as_a_full_read() -> crate::Result<()> {
use crate::table::columnar::{COL_SEQNO, COL_USER_KEY, COL_VALUE};
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file(&file, 400, 32, 96, 64 * 1_024)?;
let fresh = || Table::recover(test_recover_params(file.clone(), checksum));
let group = {
let first = fresh()?
.data_block_handles()
.next()
.expect("the table has a row group")?;
*first.as_ref()
};
let (key_at, key_len) = page_extent(&file, &group, COL_USER_KEY);
let group_at = usize::try_from(*group.offset()).expect("offset fits");
assert!(
key_at < group_at + 4_096 && key_at + key_len > group_at + 4_096,
"the fixture must put the key page across the prefix's end",
);
let full = fresh()?
.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?
.into_batch()?;
let column = |id: u16| {
full.columns
.iter()
.find(|c| c.column_id == id)
.cloned()
.expect("the full read holds every column")
};
for projection in [
&[COL_USER_KEY][..],
&[COL_VALUE][..],
&[COL_USER_KEY, COL_VALUE][..],
&[COL_SEQNO][..],
] {
let projected = fresh()?
.load_row_group(
&group,
&PageWant {
columns: Some(projection),
..PageWant::ALL
},
ReadCharge::Foreground,
)?
.into_batch()?;
assert_eq!(projected.row_count, full.row_count);
let expected: Vec<_> = full
.columns
.iter()
.filter(|c| projection.contains(&c.column_id))
.map(|c| column(c.column_id))
.collect();
assert_eq!(
projected.columns, expected,
"projection {projection:?} must decode what the full read decodes",
);
}
Ok(())
}
#[cfg(all(feature = "columnar", feature = "std"))]
#[test]
fn a_paced_projection_times_each_page_region_on_its_own() -> crate::Result<()> {
use crate::table::columnar::COL_VALUE;
use crate::table::row_group::{GroupRead, PageWant, RowPageSelect};
use crate::table::util::ReadPacer;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Timed(AtomicUsize);
impl ReadPacer for Timed {
fn active(&self) -> bool {
true
}
fn pace(&self, _offset: u64, _len: u64) {}
fn read_took(&self, _elapsed: core::time::Duration) {
self.0.fetch_add(1, Ordering::Relaxed);
}
}
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file(&file, 400, 32, 96, 64 * 1_024)?;
let fresh = || Table::recover(test_recover_params(file.clone(), checksum));
let want = PageWant::projected(&[COL_VALUE], RowPageSelect::All);
let group = {
let first = fresh()?
.data_block_handles()
.next()
.expect("the table has a row group")?;
*first.as_ref()
};
let unpaced = fresh()?
.load_row_group(&group, &want, ReadCharge::Foreground)?
.into_batch()?;
let timed = Timed(AtomicUsize::new(0));
let table = fresh()?;
let fd = crate::table::util::LookupFd::default();
let blocks = GroupRead {
pace: Some(&timed),
..table.group_read(&group, ReadCharge::Foreground, &fd)
}
.load(&want)?;
let paced = table
.row_pages(&blocks, ReadCharge::Foreground)?
.into_batch()?;
assert_eq!(paced.columns, unpaced.columns);
assert!(
timed.0.load(Ordering::Relaxed) > 0,
"the pages past the prefix are read under the pacer"
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_projection_refuses_a_directory_longer_than_its_group() -> crate::Result<()> {
use crate::table::columnar::COL_USER_KEY;
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file(&file, 200, 12, 256, 64 * 1_024)?;
let group = {
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let first = table
.data_block_handles()
.next()
.expect("the table has a row group")?;
*first.as_ref()
};
let clipped = BlockHandle::new(group.offset(), 64).with_row_group(group.row_group());
let table = Table::recover(test_recover_params(file, checksum))?;
let err = table
.load_row_group(
&clipped,
&PageWant::projected(&[COL_USER_KEY], crate::table::row_group::RowPageSelect::All),
ReadCharge::Foreground,
)
.expect_err("a directory past the group's end must be refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"expected a framing refusal, got {err:?}",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_directory_length_other_than_the_indexed_one_is_refused() -> crate::Result<()> {
use crate::table::columnar::COL_USER_KEY;
use crate::table::row_group::{PageWant, RowPageSelect};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 16 * 1_024, 1_024)?;
let group = first_row_group(&file, checksum)?;
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
assert!(
named.head_zones_len > 0,
"the fixture's groups carry a head zone block"
);
let key = b"key000010kkkkkkk";
let point = PageWant::projected(
&[COL_USER_KEY],
RowPageSelect::Zone {
column_id: COL_USER_KEY,
lower: Some(key),
upper: Some(key),
},
);
let projection = PageWant::projected(&[COL_USER_KEY], RowPageSelect::All);
let shifted = |delta: i64, of: u32| {
u32::try_from(i64::from(of) + delta).expect("a length a byte off stays a u32")
};
for (what, delta_directory, delta_head) in [
("a shorter directory", -1_i64, 0_i64),
("a longer directory", 1, 0),
("a shorter head zone block", 0, -1),
("a longer head zone block", 0, 1),
] {
let wrong = group.with_row_group(Some(crate::table::index_block::RowGroupRef {
tag: named.tag,
directory_len: core::num::NonZeroU32::new(shifted(
delta_directory,
named.directory_len.get(),
))
.expect("the directory is longer than a byte"),
head_zones_len: shifted(delta_head, named.head_zones_len),
}));
for want in [&point, &projection] {
let cold = Table::recover(test_recover_params(file.clone(), checksum))?;
let err = cold
.load_row_group(&wrong, want, ReadCharge::Foreground)
.expect_err("a cold read of a directory of another length must be refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"{what}: expected a framing refusal, got {err:?}",
);
assert!(
cold.data_unit_blocks(&wrong).is_err(),
"{what}: the walks must refuse it too",
);
let warm = Table::recover(test_recover_params(file.clone(), checksum))?;
warm.load_row_group(&group, want, ReadCharge::Foreground)?;
let err = warm
.load_row_group(&wrong, want, ReadCharge::Foreground)
.expect_err("a cached directory of another length must be refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"{what}: expected a framing refusal, got {err:?}",
);
}
}
Ok(())
}
#[cfg(all(feature = "columnar", feature = "std"))]
#[test]
fn a_paced_scan_reports_the_reads_of_its_index_walk() -> crate::Result<()> {
use crate::table::util::{Pacer, ReadPacer};
use std::sync::atomic::{AtomicUsize, Ordering};
struct Counted(AtomicUsize);
impl ReadPacer for Counted {
fn active(&self) -> bool {
true
}
fn pace(&self, _offset: u64, _len: u64) {
self.0.fetch_add(1, Ordering::Relaxed);
}
}
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 16 * 1_024, 1_024)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let counted = Arc::new(Counted(AtomicUsize::new(0)));
let pace: Pacer = counted.clone();
let rows = table.scan_paced(Some(&pace))?.count();
assert_eq!(rows, 400, "the scan reads every row");
assert!(
counted.0.load(Ordering::Relaxed) > 0,
"the index walk's reads were not paced"
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_scan_refuses_a_group_whose_lengths_disagree_with_its_index_entry() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 16 * 1_024, 1_024)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let groups = table
.data_block_handles()
.map(|keyed| keyed.map(|keyed| *keyed.as_ref()))
.collect::<crate::Result<Vec<_>>>()?;
let Some(named) = groups.first().and_then(BlockHandle::row_group) else {
panic!("a columnar index entry names its group");
};
assert!(
named.head_zones_len > 0,
"the fixture's groups carry a head zone block"
);
let scanned = |groups: Vec<BlockHandle>| {
table
.scan_groups(groups.len(), 0, groups, None)?
.collect::<crate::Result<Vec<_>>>()
};
assert_eq!(
scanned(groups.clone())?.len(),
400,
"the entries as written scan every row"
);
let shifted = |delta: i64, of: u32| {
u32::try_from(i64::from(of) + delta).expect("a length a byte off stays a u32")
};
for (what, delta_directory, delta_head) in [
("a shorter directory", -1_i64, 0_i64),
("a longer directory", 1, 0),
("a shorter head zone block", 0, -1),
("a longer head zone block", 0, 1),
] {
let mut forged = groups.clone();
let Some(first) = forged.first_mut() else {
panic!("the table has a group");
};
*first = first.with_row_group(Some(crate::table::index_block::RowGroupRef {
tag: named.tag,
directory_len: core::num::NonZeroU32::new(shifted(
delta_directory,
named.directory_len.get(),
))
.expect("the directory is longer than a byte"),
head_zones_len: shifted(delta_head, named.head_zones_len),
}));
let err = scanned(forged).expect_err("a group of other lengths must be refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"{what}: expected a framing refusal, got {err:?}",
);
}
Ok(())
}
#[cfg(feature = "columnar")]
fn first_row_group(file: &std::path::Path, checksum: Checksum) -> crate::Result<BlockHandle> {
let table = Table::recover(test_recover_params(file.to_path_buf(), checksum))?;
let first = table
.data_block_handles()
.next()
.expect("the table has a row group")?;
Ok(*first.as_ref())
}
#[cfg(feature = "columnar")]
#[test]
fn a_row_group_of_many_row_pages_reads_back_every_row() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 64 * 1_024, 1_024)?;
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let pages = table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
assert!(
pages.batches.len() > 4,
"the fixture must cut its group into row pages, got {}",
pages.batches.len(),
);
let key = |i: u32| {
let mut key = alloc::format!("key{i:06}").into_bytes();
key.resize(16, b'k');
key
};
let scanned: Vec<_> = table.iter().collect::<crate::Result<_>>()?;
assert_eq!(scanned.len(), 400, "the scan returns every row");
for (i, row) in (0..).zip(&scanned) {
assert_eq!(&*row.key.user_key, key(i).as_slice(), "row {i} in order");
assert_eq!(row.key.seqno, u64::from(i) + 1);
}
for i in 0..400 {
let key = key(i);
let got = table
.get(&key, SeqNo::MAX, hash64(&key))?
.expect("every written key is found");
assert_eq!(got.key.seqno, u64::from(i) + 1, "key {i}");
assert_eq!(&*got.value, &[b'v'; 100][..], "key {i}");
}
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_group_of_more_row_pages_than_the_directory_lists_merges_them() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 20_000, 9, 8, 4 * 1_024 * 1_024, 1)?;
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let pages = table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
assert_eq!(pages.row_count(), 20_000, "the table is one group");
assert!(
u16::try_from(pages.batches.len() * 4).is_ok(),
"the directory lists every page: {} row pages",
pages.batches.len(),
);
assert!(
pages.batches.len() > 1_000,
"merging keeps the pages as fine as the directory allows: {}",
pages.batches.len(),
);
let scanned: Vec<_> = table.iter().collect::<crate::Result<_>>()?;
assert_eq!(scanned.len(), 20_000, "the scan returns every row");
for (i, row) in (0..).zip(&scanned) {
assert_eq!(
&*row.key.user_key,
alloc::format!("key{i:06}").as_bytes(),
"row {i} in order"
);
}
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_page_size_at_the_group_size_writes_one_row_page_per_group() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_row_group_size(4_096)
.use_columnar_page_size(4_096);
for i in 0..2_000u32 {
let value = alloc::vec![b'v'; 40 + (i as usize * 37) % 300];
writer.write(InternalValue::from_components(
alloc::format!("key{i:06}").into_bytes(),
value,
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("written");
let table = Table::recover(test_recover_params(file, checksum))?;
let mut groups = 0;
for keyed in table.data_block_handles() {
let keyed = keyed?;
let pages = table.load_row_group(keyed.as_ref(), &PageWant::ALL, ReadCharge::Foreground)?;
assert_eq!(pages.batches.len(), 1, "group {groups} is one row page");
groups += 1;
}
assert!(groups > 50, "the fixture spans many groups, got {groups}");
Ok(())
}
#[cfg(feature = "columnar")]
const ROWS_BEFORE_THE_KEY: u32 = 20;
#[cfg(feature = "columnar")]
fn versioned_key_rows(versions: u64) -> Vec<InternalValue> {
let value = |s: u64| {
let mut v = alloc::format!("k{s:04}").into_bytes();
v.resize(100, b'.');
v
};
let mut rows: Vec<_> = (0..ROWS_BEFORE_THE_KEY)
.map(|i| {
InternalValue::from_components(
alloc::format!("a{i:02}").into_bytes(),
value(0),
1,
crate::ValueType::Value,
)
})
.collect();
for s in (1..=versions).rev() {
rows.push(InternalValue::from_components(
b"k".to_vec(),
value(s),
s,
crate::ValueType::Value,
));
}
rows.push(InternalValue::from_components(
b"z".to_vec(),
value(0),
1,
crate::ValueType::Value,
));
rows
}
#[cfg(feature = "columnar")]
#[test]
fn a_key_whose_versions_cross_row_pages_reads_every_version() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_row_group_size(64 * 1_024)
.use_columnar_page_size(1_024);
for row in versioned_key_rows(40) {
writer.write(row)?;
}
let (_, checksum) = writer.finish()?.expect("written");
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
assert!(
table
.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?
.batches
.len()
>= 4,
"the fixture must spread the key's versions over row pages",
);
for s in 1..=40u64 {
let got = table
.get(b"k", s + 1, hash64(b"k"))?
.expect("every version is visible at its own snapshot");
assert_eq!(got.key.seqno, s, "snapshot {}", s + 1);
assert_eq!(&got.value[..5], alloc::format!("k{s:04}").as_bytes());
}
assert!(
table.get(b"k", 1, hash64(b"k"))?.is_none(),
"before version 1"
);
assert!(table.get(b"a00", SeqNo::MAX, hash64(b"a00"))?.is_some());
assert!(table.get(b"a19", SeqNo::MAX, hash64(b"a19"))?.is_some());
assert!(table.get(b"z", SeqNo::MAX, hash64(b"z"))?.is_some());
assert!(table.get(b"m", SeqNo::MAX, hash64(b"m"))?.is_none());
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn positional_deletes_mask_the_right_rows_on_every_row_page() -> crate::Result<()> {
use crate::config::DeleteStrategy;
let dir = tempdir()?;
let file = dir.path().join("table");
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(64 * 1_024)
.use_columnar_page_size(1_024)
.delete_strategy(DeleteStrategy::MergeOnRead);
for row in versioned_key_rows(40) {
writer.write(row)?;
}
writer.delete_bitmap_mut().insert(ROWS_BEFORE_THE_KEY);
writer.delete_bitmap_mut().insert(ROWS_BEFORE_THE_KEY + 29);
let (_, checksum) = writer.finish()?.expect("written");
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let first_key_page = table
.load_row_group(
&group,
&crate::table::row_group::PageWant::ALL,
ReadCharge::Foreground,
)?
.batches
.first()
.map(|page| page.row_count)
.expect("a row page");
assert!(
first_key_page <= ROWS_BEFORE_THE_KEY,
"the key's run must start past the group's first row page",
);
let newest = table
.get(b"k", SeqNo::MAX, hash64(b"k"))?
.expect("older versions survive");
assert_eq!(newest.key.seqno, 39, "version 40 is masked");
let below_masked = table
.get(b"k", 12, hash64(b"k"))?
.expect("older versions survive");
assert_eq!(below_masked.key.seqno, 10, "version 11 is masked");
let above_masked = table
.get(b"k", 13, hash64(b"k"))?
.expect("version 12 is live");
assert_eq!(above_masked.key.seqno, 12);
let scanned: Vec<_> = table.iter().collect::<crate::Result<_>>()?;
assert_eq!(
scanned.len(),
ROWS_BEFORE_THE_KEY as usize + 39,
"every row less the two masked",
);
assert!(
scanned
.iter()
.all(|r| r.key.seqno != 40 && r.key.seqno != 11 || &*r.key.user_key != b"k"),
"the scan masks the same rows",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_page_swapped_with_another_row_page_is_refused() -> crate::Result<()> {
use crate::coding::Decode;
use crate::table::block::Header;
use crate::table::columnar::COL_SEQNO;
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 64 * 1_024, 1_024)?;
let group = first_row_group(&file, checksum)?;
let mut bytes = std::fs::read(&file)?;
let at = usize::try_from(*group.offset()).expect("offset fits");
let (directory, directory_len) = {
let frame = bytes.get(at..).expect("group within the file");
let header = Header::decode_from(&mut &frame[..])?;
let payload_at = Header::header_len(header.block_type);
let payload = frame
.get(payload_at..payload_at + header.data_length as usize)
.expect("directory payload");
(
crate::table::column_page::PageDirectory::decode(payload)?,
header.on_disk_size_with(None) as usize,
)
};
let seqno_pages: Vec<_> = directory
.entries()
.iter()
.filter(|e| e.id.column_id == COL_SEQNO)
.collect();
let (first, second) = seqno_pages
.iter()
.zip(seqno_pages.iter().skip(1))
.find(|(a, b)| a.length == b.length)
.expect("two seqno pages of one length");
let pages_at = at + directory_len + directory.pages_start() as usize;
let extent = |e: &crate::table::column_page::PageEntry| {
let start = pages_at + e.offset as usize;
start..start + e.length as usize
};
let first_bytes = bytes.get(extent(first)).expect("page").to_vec();
let second_bytes = bytes.get(extent(second)).expect("page").to_vec();
assert_ne!(first_bytes, second_bytes, "the two pages hold other rows");
let (first_at, second_at) = (extent(first).start as u64, extent(second).start as u64);
let page = bytes.get_mut(extent(first)).expect("page");
page.copy_from_slice(&second_bytes);
crate::test_forge::rebind_moved_frame(page, (0, second_at), (0, first_at))?;
let page = bytes.get_mut(extent(second)).expect("page");
page.copy_from_slice(&first_bytes);
crate::test_forge::rebind_moved_frame(page, (0, first_at), (0, second_at))?;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let err = table
.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)
.expect_err("a page in another row page's place must be refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"expected a stamp refusal, got {err:?}",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_page_refused_by_its_stamp_is_not_cached() -> crate::Result<()> {
use crate::coding::Decode;
use crate::table::block::Header;
use crate::table::columnar::COL_SEQNO;
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 64 * 1_024, 1_024)?;
let group = first_row_group(&file, checksum)?;
let original = std::fs::read(&file)?;
let mut bytes = original.clone();
let at = usize::try_from(*group.offset()).expect("offset fits");
let (directory, directory_len) = {
let frame = bytes.get(at..).expect("group within the file");
let header = Header::decode_from(&mut &frame[..])?;
let payload_at = Header::header_len(header.block_type);
let payload = frame
.get(payload_at..payload_at + header.data_length as usize)
.expect("directory payload");
(
crate::table::column_page::PageDirectory::decode(payload)?,
header.on_disk_size_with(None) as usize,
)
};
let seqno_pages: Vec<_> = directory
.entries()
.iter()
.filter(|e| e.id.column_id == COL_SEQNO)
.collect();
let (first, second) = seqno_pages
.iter()
.zip(seqno_pages.iter().skip(1))
.find(|(a, b)| a.length == b.length)
.expect("two seqno pages of one length");
let pages_at = at + directory_len + directory.pages_start() as usize;
let extent = |e: &crate::table::column_page::PageEntry| {
let start = pages_at + e.offset as usize;
start..start + e.length as usize
};
let first_bytes = bytes.get(extent(first)).expect("page").to_vec();
let second_bytes = bytes.get(extent(second)).expect("page").to_vec();
assert_ne!(first_bytes, second_bytes, "the two pages hold other rows");
let (first_at, second_at) = (extent(first).start as u64, extent(second).start as u64);
let page = bytes.get_mut(extent(first)).expect("page");
page.copy_from_slice(&second_bytes);
crate::test_forge::rebind_moved_frame(page, (0, second_at), (0, first_at))?;
let page = bytes.get_mut(extent(second)).expect("page");
page.copy_from_slice(&first_bytes);
crate::test_forge::rebind_moved_frame(page, (0, first_at), (0, second_at))?;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let err = table
.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)
.expect_err("a page in another row page's place is refused");
assert!(
matches!(err, crate::Error::InvalidHeader(_)),
"the stamp refuses it, got {err:?}",
);
std::fs::write(&file, &original)?;
let pages = table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
assert_eq!(
directory.row_count(),
pages.row_count(),
"with the right bytes back the same read is served whole",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_row_groups_blocks_are_refused_under_another_groups_tag() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 400, 16, 100, 16 * 1_024, 1_024)?;
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
assert!(
table.data_unit_blocks(&group).is_ok(),
"the group under its own tag"
);
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
let other = group.with_row_group(
named
.tag
.checked_add(1)
.map(|tag| crate::table::index_block::RowGroupRef { tag, ..named }),
);
let refused = table.data_unit_blocks(&other);
assert!(
refused.is_err(),
"the group's blocks under another tag must be refused, got {refused:?}",
);
Ok(())
}
#[cfg(feature = "columnar")]
fn two_groups_of_one_shape(
table: &Table,
with_zone_block: bool,
) -> crate::Result<(KeyedBlockHandle, KeyedBlockHandle)> {
let groups: Vec<KeyedBlockHandle> = table.data_block_handles().collect::<crate::Result<_>>()?;
let shape =
|group: &KeyedBlockHandle| -> crate::Result<(u32, Option<core::num::NonZeroU32>, u32)> {
let zone_block = if with_zone_block {
match table.data_unit_blocks(group.as_ref())?.last() {
Some(&(block, BlockType::ColumnZones)) => block.size(),
_ => panic!("a group of several row pages ends in its zone block"),
}
} else {
0
};
Ok((
group.size(),
group.as_ref().row_group().map(|g| g.directory_len),
zone_block,
))
};
let shapes = groups
.iter()
.map(shape)
.collect::<crate::Result<Vec<_>>>()?;
for (i, a) in shapes.iter().enumerate() {
if let Some(j) = (i + 1..shapes.len()).find(|&j| shapes.get(j) == Some(a)) {
return Ok((groups[i].clone(), groups[j].clone()));
}
}
panic!("the fixture holds two row groups of one shape");
}
#[cfg(feature = "columnar")]
#[test]
fn a_scan_refuses_row_groups_swapped_whole() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 2_000, 16, 100, 16 * 1_024, 1_024)?;
let (first, second) = {
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let (first, second) = two_groups_of_one_shape(&table, false)?;
(first.into_inner(), second.into_inner())
};
let mut bytes = std::fs::read(&file)?;
let extent = |group: &BlockHandle| {
let at = usize::try_from(*group.offset()).expect("offset fits");
at..at + group.size() as usize
};
let first_bytes = bytes.get(extent(&first)).expect("group").to_vec();
let second_bytes = bytes.get(extent(&second)).expect("group").to_vec();
let (first_at, second_at) = (*first.offset(), *second.offset());
let group = bytes.get_mut(extent(&first)).expect("group");
group.copy_from_slice(&second_bytes);
crate::test_forge::rebind_moved_extent(group, 0, second_at, first_at)?;
let group = bytes.get_mut(extent(&second)).expect("group");
group.copy_from_slice(&first_bytes);
crate::test_forge::rebind_moved_extent(group, 0, first_at, second_at)?;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let scanned = table
.scan()
.and_then(Iterator::collect::<crate::Result<Vec<_>>>);
assert!(
scanned.is_err(),
"a scan over swapped row groups must be refused, got {} rows",
scanned.map_or(0, |rows| rows.len()),
);
Ok(())
}
#[cfg(feature = "columnar")]
fn a_later_groups_directory_in_an_earlier(
file: &std::path::Path,
checksum: Checksum,
) -> crate::Result<(KeyedBlockHandle, Vec<u8>)> {
use crate::coding::Decode;
use crate::table::block::Header;
let (earlier, later) = {
let table = Table::recover(test_recover_params(file.to_path_buf(), checksum))?;
two_groups_of_one_shape(&table, false)?
};
let (first, second) = (*earlier.as_ref(), *later.as_ref());
let original = std::fs::read(file)?;
let mut bytes = original.clone();
let directory_at = |group: &BlockHandle| -> crate::Result<core::ops::Range<usize>> {
let at = usize::try_from(*group.offset()).expect("offset fits");
let frame = original.get(at..).expect("group within the file");
let len = Header::decode_from(&mut &frame[..])?.on_disk_size_with(None) as usize;
Ok(at..at + len)
};
let (into, from) = (directory_at(&first)?, directory_at(&second)?);
assert_eq!(
into.len(),
from.len(),
"the two directories are of one length"
);
let foreign = original.get(from.clone()).expect("directory").to_vec();
let directory = bytes.get_mut(into.clone()).expect("directory");
directory.copy_from_slice(&foreign);
crate::test_forge::rebind_moved_frame(
directory,
(0, from.start as u64),
(0, into.start as u64),
)?;
std::fs::write(file, &bytes)?;
Ok((earlier, original))
}
#[cfg(feature = "columnar")]
#[test]
fn a_point_read_refuses_another_groups_directory_whose_zones_prune_every_page() -> crate::Result<()>
{
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 2_000, 16, 100, 16 * 1_024, 1_024)?;
let (earlier, original) = a_later_groups_directory_in_an_earlier(&file, checksum)?;
let key = earlier.end_key().to_vec();
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let got = table.get(&key, crate::MAX_SEQNO, crate::hash::hash64(&key));
assert!(
got.is_err(),
"a key of the earlier group, read through the later one's directory, got {got:?}",
);
std::fs::write(&file, &original)?;
let got = table.get(&key, crate::MAX_SEQNO, crate::hash::hash64(&key))?;
assert!(got.is_some(), "with the right bytes back the key is found");
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_directory_refused_by_its_pages_is_not_cached() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = columnar_table_file_paged(&file, 2_000, 16, 100, 16 * 1_024, 1_024)?;
let (earlier, original) = a_later_groups_directory_in_an_earlier(&file, checksum)?;
let first = *earlier.as_ref();
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
assert!(
table
.load_row_group(&first, &PageWant::ALL, ReadCharge::Foreground)
.is_err(),
"another group's directory is refused by this group's pages",
);
std::fs::write(&file, &original)?;
let pages = table.load_row_group(&first, &PageWant::ALL, ReadCharge::Foreground)?;
assert!(
pages.row_count() > 0
&& (0u16..)
.zip(&pages.ordinals)
.all(|(expected, &ordinal)| expected == ordinal),
"with the right bytes back the same read is served whole: row pages {:?}",
pages.ordinals,
);
Ok(())
}
#[cfg(feature = "columnar")]
fn zoned_key(i: u32) -> Vec<u8> {
alloc::format!("key{i:06}").into_bytes()
}
#[cfg(feature = "columnar")]
fn zoned_value(i: u32) -> Vec<u8> {
let mut value = alloc::format!("val{i:06}").into_bytes();
value.resize(100, b'v');
value
}
#[cfg(feature = "columnar")]
fn zoned_table_file(file: &std::path::Path, rows: u32, deletes: &[u32]) -> crate::Result<Checksum> {
use crate::config::DeleteStrategy;
let mut writer = Writer::new(file.to_path_buf(), 0, 0, Arc::new(StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(64 * 1_024)
.use_columnar_page_size(1_024)
.delete_strategy(DeleteStrategy::MergeOnRead);
for i in 0..rows {
writer.write(InternalValue::from_components(
zoned_key(i),
zoned_value(i),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
for &pos in deletes {
writer.delete_bitmap_mut().insert(pos);
}
let (_, checksum) = writer.finish()?.expect("written");
Ok(checksum)
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
fn zoned_table_counting_reads(
file: &std::path::Path,
checksum: Checksum,
) -> crate::Result<(
Table,
Arc<crate::metrics::Metrics>,
Arc<crate::fs::FaultInjector>,
)> {
use crate::fs::FaultFs;
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
let metrics = Arc::new(crate::metrics::Metrics::default());
let mut params = test_recover_params(file.to_path_buf(), checksum);
params.fs = Arc::new(fault);
params.metrics = Arc::clone(&metrics);
params.cache = Arc::new(crate::Cache::with_capacity_bytes(64 * 1_024 * 1_024));
let table = Table::recover(params)?;
injector.clear();
Ok((table, metrics, injector))
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
fn requested(metrics: &crate::metrics::Metrics) -> (u64, u64) {
use core::sync::atomic::Ordering::Relaxed;
(
metrics.index_block_io_requested.load(Relaxed),
metrics.data_block_io_requested.load(Relaxed),
)
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
#[test]
fn a_point_read_takes_the_directory_and_the_key_zones_in_one_request() -> crate::Result<()> {
use crate::table::columnar::COL_USER_KEY;
use crate::table::row_group::{PageWant, RowPageSelect};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 1_500, &[])?;
let (table, metrics, injector) = zoned_table_counting_reads(&file, checksum)?;
let group = first_row_group(&file, checksum)?;
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
assert!(named.head_zones_len > 0, "the key zones are a head block");
let point = |key: &[u8]| -> crate::Result<crate::table::row_group::RowPages> {
table.load_row_group(
&group,
&PageWant::projected(
&[COL_USER_KEY],
RowPageSelect::Zone {
column_id: COL_USER_KEY,
lower: Some(key),
upper: Some(key),
},
),
ReadCharge::Foreground,
)
};
let (index_before, _) = requested(&metrics);
let pages = point(&zoned_key(10))?;
assert_eq!(
pages.ordinals.len(),
1,
"the key zones select the one key page holding the key",
);
assert_eq!(
requested(&metrics).0 - index_before,
u64::from(named.directory_len.get()) + u64::from(named.head_zones_len),
"the directory and the key zones, and nothing past them",
);
assert_eq!(
injector.read_count(),
2,
"one request for the directory and the key zones, one for the key page",
);
let (index_before, _) = requested(&metrics);
let reads_before = injector.read_count();
let cached_before = metrics.index_block_load_cached_count();
let pages = point(&zoned_key(200))?;
assert_eq!(pages.ordinals.len(), 1);
assert_eq!(
requested(&metrics).0,
index_before,
"the cached directory serves the key zones"
);
assert_eq!(
injector.read_count() - reads_before,
1,
"only the other key page is read",
);
assert_eq!(
metrics.index_block_load_cached_count() - cached_before,
1,
"the directory is served from the cache and no zone block is looked up",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_point_read_with_nothing_cached_reads_the_directory_once() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 1_500, &[])?;
let group = first_row_group(&file, checksum)?;
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
let mut params = test_recover_params(file, checksum);
params.fs = Arc::new(fault);
params.cache = Arc::new(crate::Cache::with_capacity_bytes(0));
let table = Table::recover(params)?;
injector.clear();
injector.arm(
FaultRule::new(FaultOp::ReadAt, Fault::Error(crate::io::ErrorKind::Other))
.at_offset(*group.offset())
.skip(1),
);
let key = zoned_key(10);
let found = table.get(&key, SeqNo::MAX, hash64(&key))?;
assert_eq!(
found.map(|value| value.value.to_vec()),
Some(zoned_value(10)),
"one read of the directory serves both steps",
);
Ok(())
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
#[test]
fn a_point_read_over_budget_reads_the_directory_and_the_key_zones_apart() -> crate::Result<()> {
use crate::table::columnar::COL_USER_KEY;
use crate::table::row_group::{PageWant, RowPageSelect};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 1_500, &[])?;
let (table, metrics, injector) = zoned_table_counting_reads(&file, checksum)?;
let group = first_row_group(&file, checksum)?;
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
assert!(named.head_zones_len > 0, "the key zones are a head block");
let budget = crate::config::ReadBudget::new(named.directory_len.get(), 16);
assert_eq!(
budget.io_buffer(),
named.directory_len.get(),
"the buffer holds exactly the directory",
);
let _ = table.0.read_budget.set(Box::new(budget));
let (index_before, _) = requested(&metrics);
let key = zoned_key(10);
let pages = table.load_row_group(
&group,
&PageWant::projected(
&[COL_USER_KEY],
RowPageSelect::Zone {
column_id: COL_USER_KEY,
lower: Some(&key),
upper: Some(&key),
},
),
ReadCharge::Foreground,
)?;
assert_eq!(
pages.ordinals.len(),
1,
"the key zones still select one page"
);
assert_eq!(
requested(&metrics).0 - index_before,
u64::from(named.directory_len.get()) + u64::from(named.head_zones_len),
"the directory and the key zones, and nothing past them",
);
assert_eq!(
injector.read_count(),
3,
"the directory, then the key zones, then the key page",
);
Ok(())
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
#[test]
fn a_read_that_does_not_select_by_the_key_never_reads_its_zones() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use crate::table::row_group::{PageWant, RowPageSelect};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 1_500, &[])?;
let group = first_row_group(&file, checksum)?;
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
let directory_len = u64::from(named.directory_len.get());
let (table, metrics, _) = zoned_table_counting_reads(&file, checksum)?;
let (index_before, _) = requested(&metrics);
table.load_row_group(
&group,
&PageWant::projected(&[COL_USER_KEY], RowPageSelect::All),
ReadCharge::Foreground,
)?;
assert_eq!(
requested(&metrics).0 - index_before,
directory_len,
"a projection reads the directory alone",
);
let (table, metrics, _) = zoned_table_counting_reads(&file, checksum)?;
let value = zoned_value(10);
let (index_before, _) = requested(&metrics);
table.load_row_group(
&group,
&PageWant::projected(
&[COL_USER_KEY, COL_VALUE],
RowPageSelect::Zone {
column_id: COL_VALUE,
lower: Some(&value),
upper: Some(&value),
},
),
ReadCharge::Foreground,
)?;
let Some((directory, _)) = table
.cache
.get_directory(table.global_id(), group.offset(), false)
else {
panic!("the read cached the directory");
};
assert!(
directory.lacks_head_zones_for(COL_USER_KEY),
"a read that did not select by the key cached no key zones",
);
let Some((_, value_zones_len)) = directory.zone_block(COL_VALUE) else {
panic!("the value column has a zone block");
};
assert_eq!(
requested(&metrics).0 - index_before,
directory_len + u64::from(value_zones_len),
"a scan pruning on the value reads the directory and the value's zone block",
);
let key = zoned_key(10);
let point = PageWant::projected(
&[COL_USER_KEY],
RowPageSelect::Zone {
column_id: COL_USER_KEY,
lower: Some(&key),
upper: Some(&key),
},
);
let (index_before, _) = requested(&metrics);
table.load_row_group(&group, &point, ReadCharge::Foreground)?;
assert_eq!(
requested(&metrics).0 - index_before,
u64::from(named.head_zones_len),
"the first point read takes the key zones alone",
);
let (index_before, _) = requested(&metrics);
table.load_row_group(&group, &point, ReadCharge::Foreground)?;
assert_eq!(
requested(&metrics).0,
index_before,
"and the cached directory keeps them for the next",
);
Ok(())
}
#[cfg(all(feature = "columnar", feature = "metrics"))]
#[test]
fn a_sparse_predicate_scan_reads_the_directory_its_zones_and_the_matching_pages()
-> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use crate::table::columnar_predicate::{ColumnRangePredicate, PredicateApply};
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 1_500, &[])?;
let group = first_row_group(&file, checksum)?;
let Some(named) = group.row_group() else {
panic!("a columnar index entry names its group");
};
let (table, metrics, _) = zoned_table_counting_reads(&file, checksum)?;
let (index_before, data_before) = requested(&metrics);
let nothing = ColumnRangePredicate {
column_id: COL_VALUE,
lower: Some(b"zzz".to_vec()),
upper: None,
apply: PredicateApply::Filter,
};
assert!(
table
.columnar_scan(&[COL_USER_KEY, COL_VALUE], Some(¬hing))?
.next()
.is_none()
);
let (index_walked, data_walked) = requested(&metrics);
assert_eq!(
data_walked, data_before,
"a group the zone map rules out is not read"
);
assert!(
table
.cache
.get_directory(table.global_id(), group.offset(), false)
.is_none(),
"nor its directory: {} index bytes were the table's index",
index_walked - index_before,
);
let row = 10;
let predicate = ColumnRangePredicate {
column_id: COL_VALUE,
lower: Some(zoned_value(row)),
upper: Some(zoned_value(row)),
apply: PredicateApply::Filter,
};
let (index_before, data_before) = requested(&metrics);
let batches = table
.columnar_scan(&[COL_USER_KEY, COL_VALUE], Some(&predicate))?
.collect::<crate::Result<Vec<_>>>()?;
let (index_after, data_after) = requested(&metrics);
assert_eq!(
batches.iter().map(|b| b.row_count).sum::<u32>(),
1,
"the scan returns the one matching row",
);
let Some((directory, _)) = table
.cache
.get_directory(table.global_id(), group.offset(), false)
else {
panic!("the scan cached the directory");
};
let Some((_, value_zones_len)) = directory.zone_block(COL_VALUE) else {
panic!("the value column has a zone block");
};
let matching_page = (0u16..)
.zip(directory.row_pages())
.find(|&(ordinal, &rows)| {
directory
.row_page_start(ordinal)
.is_some_and(|start| start <= row && row < start + rows)
})
.map(|(ordinal, _)| ordinal)
.expect("a row page holds the row");
assert!(
directory.row_pages().len() > 2,
"the group has row pages to prune"
);
let pages: u64 = directory
.entries()
.iter()
.filter(|e| {
e.row_page == matching_page && [COL_USER_KEY, COL_VALUE].contains(&e.id.column_id)
})
.map(|e| u64::from(e.length))
.sum();
assert_eq!(
index_after - index_before,
u64::from(named.directory_len.get()) + u64::from(value_zones_len),
"the directory and the predicate column's zone block",
);
assert_eq!(
data_after - data_before,
pages,
"the projected columns' pages on the matching row page",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_failed_page_read_fails_only_the_reads_that_need_the_page() -> crate::Result<()> {
use crate::encryption::EncryptionProvider;
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule};
use crate::io::ErrorKind;
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use crate::table::row_group::{PageWant, RowPageSelect};
type Variant = (
&'static str,
Option<Arc<dyn EncryptionProvider>>,
fn(Writer) -> Writer,
);
#[cfg_attr(
not(any(feature = "encryption", feature = "page_ecc")),
expect(unused_mut, reason = "the other variants are feature-gated")
)]
let mut variants: Vec<Variant> = alloc::vec![("plain", None, |w| w)];
#[cfg(feature = "encryption")]
variants.push((
"encrypted",
Some(Arc::new(crate::encryption::Aes256GcmProvider::new(
&[7u8; 32],
))),
|w| w,
));
#[cfg(feature = "page_ecc")]
variants.push(("page-ecc", None, |w| {
w.use_page_ecc(
true,
crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 4,
parity_shards: 2,
},
)
}));
for (label, encryption, configure) in variants {
let dir = tempdir()?;
let file = dir.path().join("table");
let writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_encryption(encryption.clone())
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(64 * 1_024)
.use_columnar_page_size(1_024);
let mut writer = configure(writer);
for i in 0..400 {
writer.write(InternalValue::from_components(
zoned_key(i),
zoned_value(i),
u64::from(i) + 1,
crate::ValueType::Value,
))?;
}
let (_, checksum) = writer.finish()?.expect("written");
let open = |fs: Arc<dyn crate::fs::Fs>| {
let mut params = test_recover_params(file.clone(), checksum);
params.fs = fs;
params.encryption = encryption.clone();
Table::recover(params)
};
let (group, starts, victim_at) = {
let table = open(Arc::new(StdFs))?;
let Some(Ok(group)) = table.data_block_handles().next() else {
panic!("{label}: the table has a row group");
};
let group = group.into_inner();
let pages = table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
let Some((directory, directory_len)) =
table
.cache
.get_directory(table.global_id(), group.offset(), false)
else {
panic!("{label}: the read cached the directory");
};
let Some(victim) = directory
.entries()
.iter()
.find(|e| e.id.column_id == COL_VALUE && e.row_page == 2)
else {
panic!("{label}: the group has a value page on row page 2");
};
let victim_at = *group.offset()
+ u64::from(directory_len)
+ u64::from(directory.pages_start())
+ u64::from(victim.offset);
let starts: Vec<u32> = pages
.ordinals
.iter()
.map(|&ordinal| {
directory
.row_page_start(ordinal)
.expect("a row page the read returned")
})
.collect();
(group, starts, victim_at)
};
assert!(starts.len() > 4, "{label}: the group has several row pages");
let Some((&victim_row, &other_row)) = starts.get(2).zip(starts.get(4)) else {
panic!("{label}: row pages 2 and 4 exist");
};
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
injector.arm(
FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).at_offset(victim_at),
);
let table = open(Arc::new(fault))?;
let get = |i: u32| {
let key = zoned_key(i);
table.get(&key, crate::MAX_SEQNO, crate::hash::hash64(&key))
};
let Some(other) = get(other_row)? else {
panic!("{label}: a key on another row page is found");
};
assert_eq!(
&*other.value,
zoned_value(other_row).as_slice(),
"{label}: a key on another row page reads its own value",
);
let keys_only = table.load_row_group(
&group,
&PageWant::projected(&[COL_USER_KEY], RowPageSelect::All),
ReadCharge::Foreground,
)?;
assert_eq!(
keys_only.into_batch()?.row_count,
400,
"{label}: a projection that skips the value column reads every key",
);
let refused = get(victim_row);
assert!(
matches!(&refused, Err(crate::Error::Io(e)) if e.kind() == ErrorKind::Other),
"{label}: the key whose value page fails to read gets the error, got {refused:?}",
);
injector.clear();
let Some(healed) = get(victim_row)? else {
panic!("{label}: the key is found once the page reads again");
};
assert_eq!(
&*healed.value,
zoned_value(victim_row).as_slice(),
"{label}: the failed read left nothing behind",
);
}
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_group_read_caches_its_directory_decoded() -> crate::Result<()> {
use crate::table::row_group::PageWant;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 400, &[])?;
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file, checksum))?;
table.load_row_group(&group, &PageWant::ALL, ReadCharge::Untraced)?;
assert!(
table
.cache
.get_directory(table.global_id(), group.offset(), false)
.is_none(),
"an untraced read caches no directory",
);
table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
let (cached, _) = table
.cache
.get_directory(table.global_id(), group.offset(), false)
.expect("the read cached the directory");
assert!(
!table.cache.has_block(table.global_id(), group.offset()),
"the directory is cached decoded, not as a raw block",
);
let pages = table.load_row_group(&group, &PageWant::ALL, ReadCharge::Foreground)?;
assert_eq!(
pages.row_count(),
cached.row_count(),
"a later read uses the cached directory",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_zone_block_moved_to_another_group_is_refused() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE};
use crate::table::columnar_predicate::ColumnRangePredicate;
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 6_000, &[])?;
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let zone_block = |group: &BlockHandle| -> crate::Result<BlockHandle> {
let blocks = table.data_unit_blocks(group)?;
let Some(&(block, BlockType::ColumnZones)) = blocks.last() else {
panic!("a group of several row pages ends in its zone block");
};
Ok(block)
};
let (earlier, later) = two_groups_of_one_shape(&table, true)?;
let (a, b) = (zone_block(earlier.as_ref())?, zone_block(later.as_ref())?);
let last_row: u32 = core::str::from_utf8(earlier.end_key().get(3..).expect("a zoned key"))
.ok()
.and_then(|digits| digits.parse().ok())
.expect("a zoned key ends in its row number");
drop(table);
let mut bytes = std::fs::read(&file)?;
let extent = |h: &BlockHandle| {
let start = usize::try_from(*h.offset()).expect("offset fits");
start..start + h.size() as usize
};
let a_bytes = bytes.get(extent(&a)).expect("zone block").to_vec();
let b_bytes = bytes.get(extent(&b)).expect("zone block").to_vec();
assert_ne!(a_bytes, b_bytes, "the two groups hold other values");
let (a_at, b_at) = (*a.offset(), *b.offset());
let block = bytes.get_mut(extent(&a)).expect("zone block");
block.copy_from_slice(&b_bytes);
crate::test_forge::rebind_moved_frame(block, (0, b_at), (0, a_at))?;
let block = bytes.get_mut(extent(&b)).expect("zone block");
block.copy_from_slice(&a_bytes);
crate::test_forge::rebind_moved_frame(block, (0, a_at), (0, b_at))?;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let predicate = ColumnRangePredicate {
column_id: COL_VALUE,
lower: Some(zoned_value(last_row)),
upper: Some(zoned_value(last_row)),
apply: crate::table::columnar_predicate::PredicateApply::Filter,
};
let result = table
.columnar_scan(&[COL_USER_KEY, COL_VALUE], Some(&predicate))
.and_then(Iterator::collect::<crate::Result<Vec<_>>>);
assert!(
matches!(result, Err(crate::Error::InvalidHeader(_))),
"a zone block of another group must be refused, got {:?}",
result.map(|batches| batches.iter().map(|b| b.row_count).sum::<u32>()),
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn patrol_scrub_checks_a_groups_zone_block() -> crate::Result<()> {
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, 400, &[])?;
let group = first_row_group(&file, checksum)?;
let table = Table::recover(test_recover_params(file.clone(), checksum))?;
let blocks = table.data_unit_blocks(&group)?;
let Some(&(zone_block, role)) = blocks.last() else {
panic!("a group has blocks");
};
assert_eq!(
role,
BlockType::ColumnZones,
"the group ends in its zone block"
);
let clean = table.scrub_data_blocks();
assert_eq!(clean.uncorrectable_blocks, 0, "a clean table scrubs clean");
assert!(
clean.blocks_scanned >= blocks.len(),
"every block of the group is scanned, the zone block included",
);
drop(table);
let mut bytes = std::fs::read(&file)?;
let at = usize::try_from(*zone_block.offset()).expect("offset fits")
+ zone_block.size() as usize
- 1;
*bytes.get_mut(at).expect("byte within the zone block") ^= 0x40;
std::fs::write(&file, &bytes)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let damaged = table.scrub_data_blocks();
assert_eq!(
damaged.uncorrectable_blocks, 1,
"the damaged zone block is reported: {damaged:?}",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn a_predicate_scan_over_row_pages_returns_what_filtering_every_row_returns() -> crate::Result<()> {
use crate::table::columnar::{COL_USER_KEY, COL_VALUE, bytes_column_row};
use crate::table::columnar_predicate::ColumnRangePredicate;
const ROWS: u32 = 400;
const DELETES: [u32; 3] = [150, 151, 205];
let dir = tempdir()?;
let file = dir.path().join("table");
let checksum = zoned_table_file(&file, ROWS, &DELETES)?;
let table = Table::recover(test_recover_params(file, checksum))?;
let cases = [
(COL_USER_KEY, zoned_key(140), zoned_key(160)),
(COL_VALUE, zoned_value(200), zoned_value(209)),
(COL_USER_KEY, zoned_key(0), zoned_key(0)),
(COL_VALUE, zoned_value(ROWS - 1), zoned_value(ROWS - 1)),
(COL_USER_KEY, zoned_key(ROWS), zoned_key(ROWS + 10)),
];
for (column_id, lower, upper) in cases {
let predicate = ColumnRangePredicate {
column_id,
lower: Some(lower.clone()),
upper: Some(upper.clone()),
apply: crate::table::columnar_predicate::PredicateApply::Filter,
};
let mut got = Vec::new();
for batch in table.columnar_scan(&[COL_USER_KEY, COL_VALUE], Some(&predicate))? {
let batch = batch?;
let keys = batch
.columns
.iter()
.find(|c| c.column_id == COL_USER_KEY)
.expect("the key column");
for row in 0..batch.row_count {
got.push(bytes_column_row(&keys.data, batch.row_count, row)?.to_vec());
}
}
let expected: Vec<Vec<u8>> = (0..ROWS)
.filter(|i| !DELETES.contains(i))
.filter(|&i| {
let cell = if column_id == COL_USER_KEY {
zoned_key(i)
} else {
zoned_value(i)
};
lower <= cell && cell <= upper
})
.map(zoned_key)
.collect();
assert_eq!(
got, expected,
"column {column_id} in {lower:?}..={upper:?} returns what a filter of every row does",
);
}
Ok(())
}
#[test]
fn a_restricted_view_keeps_its_prefix_filter() -> crate::Result<()> {
struct FirstFour;
impl crate::prefix::PrefixExtractor for FirstFour {
fn name(&self) -> &'static str {
"fixed-4"
}
fn prefixes<'a>(&self, key: &'a [u8]) -> Box<dyn Iterator<Item = &'a [u8]> + 'a> {
Box::new(key.get(..4).into_iter())
}
}
let dir = tempdir()?;
let file = dir.path().join("0");
let extractor: Arc<dyn crate::prefix::PrefixExtractor> = Arc::new(FirstFour);
let mut writer = Writer::new(file.clone(), 0, 0, Arc::new(StdFs))?
.use_data_block_size(128)
.use_prefix_extractor(Some(Arc::clone(&extractor)));
for i in 0u64..200 {
writer.write(InternalValue::from_components(
format!("key{i:05}").into_bytes(),
b"v".to_vec(),
i + 1,
crate::ValueType::Value,
))?;
}
let checksum = writer.finish()?.expect("the SST is non-empty").1;
let mut params = test_recover_params(file, checksum);
params.prefix_extractor = Some(extractor);
let table = Table::recover(params)?;
assert!(table.has_prefix_filter());
let restricted = table.reopen_restricted(crate::UserKey::from(&b"key00100"[..]))?;
assert!(
restricted.has_prefix_filter(),
"the restricted view asks its filter as the table did"
);
Ok(())
}