#![cfg(all(feature = "typed-tree", feature = "armour", feature = "rapira-codec"))]
use std::collections::HashSet;
use std::num::NonZeroU32;
use std::ops::Bound;
use std::path::Path;
#[cfg(feature = "rpc")]
use armdb::armour::RpcHandler;
use armdb::armour::{
CountMode, Db, InspectionEncoding, InspectionError, InspectionOperation, InspectionPage,
IterationDirection, ScanCapability,
};
use armdb::{CollectionMeta, Config, DbError, FixedConfig, NoHook, RapiraCodec, TypedTree};
use armour_core::{CollectionShape, GetType, StorageBackend, StorageClass, ValueModel};
use rapira::Rapira;
use tempfile::tempdir;
use zerocopy::IntoBytes;
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspTypedTree {
n: u64,
}
impl CollectionMeta for InspTypedTree {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_typed_tree";
const VERSION: u16 = 11;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspTypedMapOrd {
n: u64,
}
impl CollectionMeta for InspTypedMapOrd {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_typed_map_ord";
const VERSION: u16 = 12;
}
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspTypedMapSnap {
n: u64,
}
impl CollectionMeta for InspTypedMapSnap {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_typed_map_snap";
const VERSION: u16 = 13;
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
)]
#[repr(C)]
struct InspZero(u32);
impl CollectionMeta for InspZero {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_zero_tree_bc";
const VERSION: u16 = 14;
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
)]
#[repr(C)]
struct InspZeroFx {
n: u32,
}
impl CollectionMeta for InspZeroFx {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_zero_tree_fx";
const VERSION: u16 = 15;
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
)]
#[repr(C)]
struct InspZeroMapOrd(InspZero);
impl CollectionMeta for InspZeroMapOrd {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_zero_map_ord";
const VERSION: u16 = 16;
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
)]
#[repr(C)]
struct InspZeroMapSnapBc(InspZero);
impl CollectionMeta for InspZeroMapSnapBc {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_zero_map_snap_bc";
const VERSION: u16 = 17;
}
#[derive(
Clone,
Copy,
Debug,
PartialEq,
GetType,
zerocopy::FromBytes,
zerocopy::IntoBytes,
zerocopy::Immutable,
)]
#[repr(C)]
struct InspZeroMapSnapFx(InspZero);
impl CollectionMeta for InspZeroMapSnapFx {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_zero_map_snap_fx";
const VERSION: u16 = 18;
}
#[cfg(feature = "var-collections")]
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspVarTree {
n: u64,
}
#[cfg(feature = "var-collections")]
impl CollectionMeta for InspVarTree {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_var_tree";
const VERSION: u16 = 19;
}
#[cfg(feature = "var-collections")]
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspVarMapOrd {
n: u64,
}
#[cfg(feature = "var-collections")]
impl CollectionMeta for InspVarMapOrd {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_var_map_ord";
const VERSION: u16 = 20;
}
#[cfg(feature = "var-collections")]
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspVarMapSnap {
n: u64,
}
#[cfg(feature = "var-collections")]
impl CollectionMeta for InspVarMapSnap {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_var_map_snap";
const VERSION: u16 = 21;
}
#[cfg(feature = "fjall")]
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspLsmTree {
n: u64,
}
#[cfg(feature = "fjall")]
impl CollectionMeta for InspLsmTree {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_lsm_tree";
const VERSION: u16 = 22;
}
#[cfg(feature = "fjall")]
#[derive(Clone, Debug, PartialEq, Rapira, GetType)]
struct InspLsmTreeReversed {
n: u64,
}
#[cfg(feature = "fjall")]
impl CollectionMeta for InspLsmTreeReversed {
type SelfId = [u8; 8];
const NAME: &'static str = "insp_lsm_tree_reversed";
const VERSION: u16 = 23;
}
fn key(id: u64) -> [u8; 8] {
id.to_be_bytes()
}
fn key_vec(id: u64) -> Vec<u8> {
key(id).to_vec()
}
fn page_keys(page: &InspectionPage) -> Vec<u64> {
page.entries
.iter()
.map(|entry| u64::from_be_bytes(entry.key.as_slice().try_into().unwrap()))
.collect()
}
fn page_key_bytes(page: &InspectionPage) -> Vec<Vec<u8>> {
page.entries.iter().map(|entry| entry.key.clone()).collect()
}
fn prefixed_key(byte0: u8, byte1: u8, last: u8) -> [u8; 8] {
let mut k = [0u8; 8];
k[0] = byte0;
k[1] = byte1;
k[7] = last;
k
}
fn encode_rapira<T: Rapira>(value: &T) -> Vec<u8> {
let codec = RapiraCodec;
let mut buf = Vec::new();
armdb::Codec::encode_to(&codec, value, &mut buf).unwrap();
buf
}
fn encode_zero<T: IntoBytes + zerocopy::Immutable>(value: &T) -> Vec<u8> {
value.as_bytes().to_vec()
}
fn asc_config() -> Config {
Config::balanced()
.shard_count(2)
.hints(true)
.iterable(true)
.reversed(false)
.build()
}
fn desc_config() -> Config {
Config::balanced()
.shard_count(2)
.hints(true)
.iterable(true)
.reversed(true)
.build()
}
fn snap_config() -> Config {
Config::balanced()
.shard_count(2)
.hints(true)
.iterable(false)
.build()
}
fn fixed_iterable(reversed: bool) -> FixedConfig {
FixedConfig {
shard_count: 2,
grow_step: 32,
iterable: true,
reversed,
..FixedConfig::test()
}
}
fn fixed_snap() -> FixedConfig {
FixedConfig {
iterable: false,
..FixedConfig::test()
}
}
const ZERO_V: usize = std::mem::size_of::<InspZero>();
#[cfg_attr(not(feature = "fjall"), allow(unused_variables))]
fn open_matrix_db(dir: &Path, include_lsm: bool) -> Db {
let db = Db::open_test(dir).expect("open db");
let tree = db
.open_typed_tree::<InspTypedTree, RapiraCodec, _>(asc_config(), NoHook, &[])
.expect("typed tree");
for id in 1..=5 {
tree.put(&key(id), InspTypedTree { n: id })
.expect("seed typed tree");
}
let map_ord = db
.open_typed_map::<InspTypedMapOrd, RapiraCodec, _>(desc_config(), NoHook, &[])
.expect("typed map ord");
for id in 1..=5 {
map_ord
.put(&key(id), InspTypedMapOrd { n: id })
.expect("seed typed map ord");
}
let map_snap = db
.open_typed_map::<InspTypedMapSnap, RapiraCodec, _>(snap_config(), NoHook, &[])
.expect("typed map snap");
for id in 1..=3 {
map_snap
.put(&key(id), InspTypedMapSnap { n: id })
.expect("seed typed map snap");
}
let ztree_bc = db
.open_zero_tree::<InspZero, ZERO_V, _>(asc_config(), NoHook, &[])
.expect("zero tree bc");
for id in 1..=5 {
ztree_bc
.put(&key(id), &InspZero(id as u32))
.expect("seed zero tree bc");
}
let ztree_fx = db
.open_zero_tree_fixed::<InspZeroFx, ZERO_V, _>(fixed_iterable(true), NoHook, &[])
.expect("zero tree fx");
for id in 1..=5 {
ztree_fx
.put(&key(id), &InspZeroFx { n: id as u32 })
.expect("seed zero tree fx");
}
let zmap_ord = db
.open_zero_map::<InspZeroMapOrd, ZERO_V, _>(asc_config(), NoHook, &[])
.expect("zero map ord");
for id in 1..=5 {
zmap_ord
.put(&key(id), &InspZeroMapOrd(InspZero(id as u32)))
.expect("seed zero map ord");
}
let zmap_snap_bc = db
.open_zero_map::<InspZeroMapSnapBc, ZERO_V, _>(snap_config(), NoHook, &[])
.expect("zero map snap bc");
for id in 1..=3 {
zmap_snap_bc
.put(&key(id), &InspZeroMapSnapBc(InspZero(id as u32)))
.expect("seed zero map snap bc");
}
let zmap_snap_fx = db
.open_zero_map_fixed::<InspZeroMapSnapFx, ZERO_V, _>(fixed_snap(), NoHook, &[])
.expect("zero map snap fx");
for id in 1..=3 {
zmap_snap_fx
.put(&key(id), &InspZeroMapSnapFx(InspZero(id as u32)))
.expect("seed zero map snap fx");
}
#[cfg(feature = "var-collections")]
{
let vtree = db
.open_var_typed_tree::<InspVarTree, RapiraCodec, _>(asc_config(), NoHook, &[])
.expect("var tree");
for id in 1..=5 {
vtree
.put(&key(id), &InspVarTree { n: id })
.expect("seed var tree");
}
let vmap_ord = db
.open_var_typed_map::<InspVarMapOrd, RapiraCodec, _>(desc_config(), NoHook, &[])
.expect("var map ord");
for id in 1..=5 {
vmap_ord
.put(&key(id), &InspVarMapOrd { n: id })
.expect("seed var map ord");
}
let vmap_snap = db
.open_var_typed_map::<InspVarMapSnap, RapiraCodec, _>(snap_config(), NoHook, &[])
.expect("var map snap");
for id in 1..=3 {
vmap_snap
.put(&key(id), &InspVarMapSnap { n: id })
.expect("seed var map snap");
}
}
#[cfg(feature = "fjall")]
if include_lsm {
use armdb::lsm::{LsmConfig, LsmDbConfig};
let lsm_cfg = LsmConfig {
reversed: false,
..LsmConfig::default()
};
let lsm = db
.open_lsm_tree::<InspLsmTree, RapiraCodec, _>(
Some(LsmDbConfig::default()),
lsm_cfg,
NoHook,
&[],
)
.expect("lsm tree");
for id in 1..=5 {
lsm.put(&key(id), &InspLsmTree { n: id })
.expect("seed lsm tree");
}
let reversed_lsm = db
.open_lsm_tree::<InspLsmTreeReversed, RapiraCodec, _>(
None,
LsmConfig {
reversed: true,
..LsmConfig::default()
},
NoHook,
&[],
)
.expect("reversed lsm tree");
for id in 1..=5 {
reversed_lsm
.put(&key(id), &InspLsmTreeReversed { n: id })
.expect("seed reversed lsm tree");
}
}
db
}
fn expected_base_names() -> Vec<&'static str> {
let mut names = vec![
InspTypedTree::NAME,
InspTypedMapOrd::NAME,
InspTypedMapSnap::NAME,
InspZero::NAME,
InspZeroFx::NAME,
InspZeroMapOrd::NAME,
InspZeroMapSnapBc::NAME,
InspZeroMapSnapFx::NAME,
];
#[cfg(feature = "var-collections")]
{
names.push(InspVarTree::NAME);
names.push(InspVarMapOrd::NAME);
names.push(InspVarMapSnap::NAME);
}
names.sort();
names
}
fn assert_invalid_key(inspector: &armdb::armour::Inspector<'_>, name: &str) {
for bad in [&[1u8; 7][..], &[1u8; 9][..]] {
let err = inspector.get(name, bad).unwrap_err();
assert!(
matches!(err, InspectionError::InvalidKey { expected: 8, .. }),
"{name}: {err:?}"
);
}
}
#[test]
fn inspection_discovery_lists_registered_collections_sorted() {
let _module_path = armdb::armour::inspection::InspectionEncoding::Opaque;
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
let direct_dir = tempdir().expect("direct tmp");
let _direct = TypedTree::<[u8; 8], InspTypedTree, RapiraCodec>::open(
direct_dir.path(),
Config::test(),
RapiraCodec,
)
.expect("direct tree");
_direct
.put(&key(99), InspTypedTree { n: 99 })
.expect("direct put");
let inspector = db.inspector();
let collections = inspector.collections().expect("collections");
let names: Vec<_> = collections.iter().map(|c| c.name.as_str()).collect();
let mut expected = expected_base_names();
expected.sort();
assert_eq!(names, expected);
let typed = collections
.iter()
.find(|c| c.name == InspTypedTree::NAME)
.expect("typed tree meta");
assert_eq!(typed.schema.version, InspTypedTree::VERSION);
assert_eq!(typed.schema.typ, InspTypedTree::TYPE);
assert_eq!(typed.schema.key_scheme, <[u8; 8] as armdb::Key>::KEY_SCHEME);
assert_eq!(typed.schema.key_size, 8);
assert_eq!(typed.schema.encoding, InspectionEncoding::Rapira);
assert_eq!(
typed.storage,
StorageClass {
model: ValueModel::Typed,
shape: CollectionShape::Tree,
backend: StorageBackend::Bitcask,
}
);
assert_eq!(typed.count, Some(5));
assert_eq!(typed.scan, ScanCapability::Ordered);
assert_eq!(typed.direction, Some(IterationDirection::Ascending));
let typed_map_ord = collections
.iter()
.find(|c| c.name == InspTypedMapOrd::NAME)
.expect("typed map ord");
assert_eq!(typed_map_ord.scan, ScanCapability::Ordered);
assert_eq!(
typed_map_ord.direction,
Some(IterationDirection::Descending)
);
assert_eq!(typed_map_ord.storage.shape, CollectionShape::Map);
let typed_map_snap = collections
.iter()
.find(|c| c.name == InspTypedMapSnap::NAME)
.expect("typed map snap");
assert_eq!(typed_map_snap.scan, ScanCapability::UnorderedSnapshot);
assert!(typed_map_snap.direction.is_none());
let zero_tree = collections
.iter()
.find(|c| c.name == InspZero::NAME)
.expect("zero tree bc");
assert_eq!(zero_tree.schema.encoding, InspectionEncoding::RawLayout);
assert_eq!(zero_tree.storage.model, ValueModel::Zero);
assert_eq!(zero_tree.storage.backend, StorageBackend::Bitcask);
let zero_tree_fx = collections
.iter()
.find(|c| c.name == InspZeroFx::NAME)
.expect("zero tree fx");
assert_eq!(zero_tree_fx.storage.backend, StorageBackend::Fixed);
assert_eq!(zero_tree_fx.direction, Some(IterationDirection::Descending));
let zero_map_snap_fx = collections
.iter()
.find(|c| c.name == InspZeroMapSnapFx::NAME)
.expect("zero map snap fx");
assert_eq!(zero_map_snap_fx.scan, ScanCapability::UnorderedSnapshot);
assert_eq!(zero_map_snap_fx.storage.backend, StorageBackend::Fixed);
#[cfg(feature = "var-collections")]
{
let var_tree = collections
.iter()
.find(|c| c.name == InspVarTree::NAME)
.expect("var tree");
assert_eq!(var_tree.schema.encoding, InspectionEncoding::Rapira);
assert_eq!(var_tree.storage.model, ValueModel::VarTyped);
assert_eq!(var_tree.scan, ScanCapability::Ordered);
}
}
#[cfg(feature = "fjall")]
#[test]
fn inspection_discovery_includes_lsm_when_enabled() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), true);
let names: Vec<_> = db
.inspector()
.collections()
.expect("collections")
.into_iter()
.map(|c| c.name)
.collect();
assert!(names.contains(&InspLsmTree::NAME.to_owned()));
assert!(names.contains(&InspLsmTreeReversed::NAME.to_owned()));
let lsm = db
.inspector()
.collections()
.expect("collections")
.into_iter()
.find(|c| c.name == InspLsmTree::NAME)
.expect("lsm meta");
assert_eq!(lsm.schema.encoding, InspectionEncoding::Rapira);
assert_eq!(lsm.storage.backend, StorageBackend::Lsm);
assert_eq!(lsm.scan, ScanCapability::Ordered);
assert_eq!(lsm.direction, Some(IterationDirection::Ascending));
let reversed_lsm = db
.inspector()
.collections()
.expect("collections")
.into_iter()
.find(|c| c.name == InspLsmTreeReversed::NAME)
.expect("reversed lsm meta");
assert_eq!(reversed_lsm.direction, Some(IterationDirection::Descending));
}
#[test]
fn inspection_exact_get_and_count_matrix() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
let inspector = db.inspector();
struct Case<'a> {
name: &'a str,
present_key: u64,
value: Vec<u8>,
count: u64,
rapira: bool,
}
let cases = [
Case {
name: InspTypedTree::NAME,
present_key: 2,
value: encode_rapira(&InspTypedTree { n: 2 }),
count: 5,
rapira: true,
},
Case {
name: InspTypedMapOrd::NAME,
present_key: 3,
value: encode_rapira(&InspTypedMapOrd { n: 3 }),
count: 5,
rapira: true,
},
Case {
name: InspTypedMapSnap::NAME,
present_key: 1,
value: encode_rapira(&InspTypedMapSnap { n: 1 }),
count: 3,
rapira: true,
},
Case {
name: InspZero::NAME,
present_key: 4,
value: encode_zero(&InspZero(4)),
count: 5,
rapira: false,
},
Case {
name: InspZeroFx::NAME,
present_key: 5,
value: encode_zero(&InspZeroFx { n: 5 }),
count: 5,
rapira: false,
},
Case {
name: InspZeroMapOrd::NAME,
present_key: 2,
value: encode_zero(&InspZeroMapOrd(InspZero(2))),
count: 5,
rapira: false,
},
Case {
name: InspZeroMapSnapBc::NAME,
present_key: 2,
value: encode_zero(&InspZeroMapSnapBc(InspZero(2))),
count: 3,
rapira: false,
},
Case {
name: InspZeroMapSnapFx::NAME,
present_key: 3,
value: encode_zero(&InspZeroMapSnapFx(InspZero(3))),
count: 3,
rapira: false,
},
];
for case in cases {
let schema = inspector.schema(case.name).expect("schema");
if case.rapira {
assert_eq!(schema.encoding, InspectionEncoding::Rapira);
} else {
assert_eq!(schema.encoding, InspectionEncoding::RawLayout);
}
let entry = inspector
.get(case.name, &key(case.present_key))
.expect("get")
.expect("present");
assert_eq!(entry.key, key_vec(case.present_key));
assert_eq!(entry.value, case.value);
if case.rapira {
armour_core::dyn_value::decode(&schema.typ, &entry.value).expect("rapira decode");
}
assert!(
inspector
.get(case.name, &key(99))
.expect("absent")
.is_none()
);
assert_invalid_key(&inspector, case.name);
assert_eq!(
inspector
.count(case.name, CountMode::Current)
.expect("current"),
case.count
);
assert_eq!(
inspector.count(case.name, CountMode::Exact).expect("exact"),
case.count
);
}
#[cfg(feature = "var-collections")]
{
for (name, present, value, count) in [
(
InspVarTree::NAME,
2u64,
encode_rapira(&InspVarTree { n: 2 }),
5u64,
),
(
InspVarMapOrd::NAME,
4,
encode_rapira(&InspVarMapOrd { n: 4 }),
5,
),
(
InspVarMapSnap::NAME,
2,
encode_rapira(&InspVarMapSnap { n: 2 }),
3,
),
] {
let schema = inspector.schema(name).expect("schema");
assert_eq!(schema.encoding, InspectionEncoding::Rapira);
let entry = inspector.get(name, &key(present)).expect("get").unwrap();
assert_eq!(entry.value, value);
armour_core::dyn_value::decode(&schema.typ, &entry.value).expect("decode");
assert_eq!(inspector.count(name, CountMode::Exact).unwrap(), count);
assert_invalid_key(&inspector, name);
}
}
}
#[cfg(feature = "rpc")]
fn insert_prefix_fixture_rows(db: &Db, name: &str) {
use armour_rpc::protocol::UpsertKey;
let handler = db
.rpc_handler_for(name)
.expect("rpc handler for prefix seed");
let trailing_1 = prefixed_key(0x01, 0xFF, 1);
let trailing_2 = prefixed_key(0x01, 0xFF, 2);
let all_ff = [0xFFu8; 8];
let upsert = |key: [u8; 8], value: Vec<u8>| {
handler
.upsert(UpsertKey::Provided(key.to_vec()), None, value)
.expect("prefix fixture upsert");
};
match name {
InspTypedTree::NAME => {
upsert(trailing_1, encode_rapira(&InspTypedTree { n: 101 }));
upsert(trailing_2, encode_rapira(&InspTypedTree { n: 102 }));
upsert(all_ff, encode_rapira(&InspTypedTree { n: 255 }));
}
InspTypedMapOrd::NAME => {
upsert(trailing_1, encode_rapira(&InspTypedMapOrd { n: 101 }));
upsert(trailing_2, encode_rapira(&InspTypedMapOrd { n: 102 }));
upsert(all_ff, encode_rapira(&InspTypedMapOrd { n: 255 }));
}
InspZero::NAME => {
upsert(trailing_1, encode_zero(&InspZero(101)));
upsert(trailing_2, encode_zero(&InspZero(102)));
upsert(all_ff, encode_zero(&InspZero(255)));
}
InspZeroFx::NAME => {
upsert(trailing_1, encode_zero(&InspZeroFx { n: 101 }));
upsert(trailing_2, encode_zero(&InspZeroFx { n: 102 }));
upsert(all_ff, encode_zero(&InspZeroFx { n: 255 }));
}
InspZeroMapOrd::NAME => {
upsert(trailing_1, encode_zero(&InspZeroMapOrd(InspZero(101))));
upsert(trailing_2, encode_zero(&InspZeroMapOrd(InspZero(102))));
upsert(all_ff, encode_zero(&InspZeroMapOrd(InspZero(255))));
}
#[cfg(feature = "var-collections")]
InspVarTree::NAME => {
upsert(trailing_1, encode_rapira(&InspVarTree { n: 101 }));
upsert(trailing_2, encode_rapira(&InspVarTree { n: 102 }));
upsert(all_ff, encode_rapira(&InspVarTree { n: 255 }));
}
#[cfg(feature = "var-collections")]
InspVarMapOrd::NAME => {
upsert(trailing_1, encode_rapira(&InspVarMapOrd { n: 101 }));
upsert(trailing_2, encode_rapira(&InspVarMapOrd { n: 102 }));
upsert(all_ff, encode_rapira(&InspVarMapOrd { n: 255 }));
}
#[cfg(feature = "fjall")]
InspLsmTree::NAME => {
upsert(trailing_1, encode_rapira(&InspLsmTree { n: 101 }));
upsert(trailing_2, encode_rapira(&InspLsmTree { n: 102 }));
upsert(all_ff, encode_rapira(&InspLsmTree { n: 255 }));
}
#[cfg(feature = "fjall")]
InspLsmTreeReversed::NAME => {
upsert(trailing_1, encode_rapira(&InspLsmTreeReversed { n: 101 }));
upsert(trailing_2, encode_rapira(&InspLsmTreeReversed { n: 102 }));
upsert(all_ff, encode_rapira(&InspLsmTreeReversed { n: 255 }));
}
other => panic!("no prefix fixture encoder for {other}"),
}
}
fn assert_inspection_scan_contract(db: &Db, name: &str, reversed: bool) {
let inspector = db.inspector();
let limit = NonZeroU32::new(10).unwrap();
let full = inspector
.range(name, Bound::Unbounded, Bound::Unbounded, None, limit)
.expect("full range");
let keys = page_keys(&full);
let expected: Vec<u64> = if reversed {
(1..=5).rev().collect()
} else {
(1..=5).collect()
};
assert_eq!(keys, expected, "{name} full range");
let mid = inspector
.range(
name,
Bound::Included(&key_vec(2)),
Bound::Excluded(&key_vec(5)),
None,
limit,
)
.expect("included/excluded range");
let mid_keys = page_keys(&mid);
let expected_mid: Vec<u64> = if reversed {
vec![4, 3, 2]
} else {
vec![2, 3, 4]
};
assert_eq!(mid_keys, expected_mid, "{name} bounded range");
let singleton = inspector
.range(
name,
Bound::Included(&key_vec(3)),
Bound::Included(&key_vec(3)),
None,
limit,
)
.expect("singleton");
assert_eq!(page_keys(&singleton), vec![3], "{name} singleton");
let empty = inspector
.range(
name,
Bound::Included(&key_vec(4)),
Bound::Excluded(&key_vec(4)),
None,
limit,
)
.expect("empty range");
assert!(empty.entries.is_empty(), "{name} equal empty");
let err = inspector
.range(
name,
Bound::Included(&key_vec(9)),
Bound::Included(&key_vec(1)),
None,
limit,
)
.unwrap_err();
assert!(
matches!(err, InspectionError::InvalidBounds { .. }),
"{name}"
);
let prefix_page = inspector
.prefix(name, &[], None, limit)
.expect("empty prefix");
assert_eq!(page_keys(&prefix_page), expected, "{name} empty prefix");
let ordinary = inspector
.prefix(name, &key_vec(1)[..1], None, limit)
.expect("ordinary prefix");
assert_eq!(page_keys(&ordinary), expected, "{name} ordinary prefix");
#[cfg(feature = "rpc")]
{
insert_prefix_fixture_rows(db, name);
let trailing_ff_prefix = [0x01u8, 0xFF];
let trailing_ff_page = inspector
.prefix(name, &trailing_ff_prefix, None, limit)
.expect("trailing ff prefix");
let mut trailing_ff_expected = vec![
prefixed_key(0x01, 0xFF, 1).to_vec(),
prefixed_key(0x01, 0xFF, 2).to_vec(),
];
if reversed {
trailing_ff_expected.reverse();
}
assert_eq!(
page_key_bytes(&trailing_ff_page),
trailing_ff_expected,
"{name} trailing ff prefix"
);
let all_ff = [0xFFu8; 8];
let all_ff_page = inspector
.prefix(name, &all_ff, None, limit)
.expect("all ff prefix");
assert_eq!(
page_key_bytes(&all_ff_page),
vec![all_ff.to_vec()],
"{name} all ff prefix"
);
}
let err = inspector.prefix(name, &[0u8; 9], None, limit).unwrap_err();
assert!(
matches!(err, InspectionError::InvalidBounds { .. }),
"{name}"
);
let err = inspector
.range(
name,
Bound::Included(&[1u8; 7]),
Bound::Unbounded,
None,
limit,
)
.unwrap_err();
assert!(matches!(err, InspectionError::InvalidKey { .. }), "{name}");
let err = inspector
.range(
name,
Bound::Unbounded,
Bound::Unbounded,
Some(&[1u8; 7]),
limit,
)
.unwrap_err();
assert!(matches!(err, InspectionError::InvalidKey { .. }), "{name}");
}
#[test]
fn inspection_ordered_range_and_prefix_matrix() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
assert_inspection_scan_contract(&db, InspTypedTree::NAME, false);
assert_inspection_scan_contract(&db, InspZero::NAME, false);
assert_inspection_scan_contract(&db, InspZeroFx::NAME, true);
assert_inspection_scan_contract(&db, InspTypedMapOrd::NAME, true);
assert_inspection_scan_contract(&db, InspZeroMapOrd::NAME, false);
#[cfg(feature = "var-collections")]
{
assert_inspection_scan_contract(&db, InspVarTree::NAME, false);
assert_inspection_scan_contract(&db, InspVarMapOrd::NAME, true);
}
}
#[cfg(feature = "fjall")]
fn assert_range_and_prefix_pagination(
inspector: &armdb::armour::Inspector<'_>,
name: &str,
reversed: bool,
) {
let expected = if reversed {
[vec![5, 4], vec![3, 2], vec![1]]
} else {
[vec![1, 2], vec![3, 4], vec![5]]
};
let limit = NonZeroU32::new(2).unwrap();
for prefix in [false, true] {
let mut after = None;
for (index, expected_keys) in expected.iter().enumerate() {
let page = if prefix {
inspector
.prefix(name, &[], after.as_deref(), limit)
.expect("prefix page")
} else {
inspector
.range(
name,
Bound::Unbounded,
Bound::Unbounded,
after.as_deref(),
limit,
)
.expect("range page")
};
assert_eq!(page_keys(&page), *expected_keys, "{name}");
let expected_cursor = (index + 1 < expected.len())
.then(|| key_vec(*expected_keys.last().expect("non-empty page")));
assert_eq!(page.next_key, expected_cursor, "{name}");
after = page.next_key;
}
}
}
#[cfg(feature = "fjall")]
#[test]
fn inspection_lsm_range_prefix_and_pagination_in_both_directions() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), true);
let inspector = db.inspector();
assert_range_and_prefix_pagination(&inspector, InspLsmTree::NAME, false);
assert_range_and_prefix_pagination(&inspector, InspLsmTreeReversed::NAME, true);
assert_inspection_scan_contract(&db, InspLsmTree::NAME, false);
assert_inspection_scan_contract(&db, InspLsmTreeReversed::NAME, true);
}
#[test]
fn inspection_pagination_contract() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
let inspector = db.inspector();
let name = InspTypedTree::NAME;
let limit2 = NonZeroU32::new(2).unwrap();
let p1 = inspector
.range(name, Bound::Unbounded, Bound::Unbounded, None, limit2)
.expect("page 1");
assert_eq!(page_keys(&p1), vec![1, 2]);
assert_eq!(p1.next_key, Some(key_vec(2)));
let p2 = inspector
.range(
name,
Bound::Unbounded,
Bound::Unbounded,
p1.next_key.as_deref(),
limit2,
)
.expect("page 2");
assert_eq!(page_keys(&p2), vec![3, 4]);
assert_eq!(p2.next_key, Some(key_vec(4)));
let p3 = inspector
.range(
name,
Bound::Unbounded,
Bound::Unbounded,
p2.next_key.as_deref(),
limit2,
)
.expect("page 3");
assert_eq!(page_keys(&p3), vec![5]);
assert!(p3.next_key.is_none());
let tiny_dir = tempdir().expect("tiny");
let tiny_db = Db::open_test(tiny_dir.path()).expect("tiny open");
let tiny = tiny_db
.open_typed_tree::<InspTypedTree, RapiraCodec, _>(asc_config(), NoHook, &[])
.expect("tiny tree");
tiny.put(&key(1), InspTypedTree { n: 1 }).expect("put");
tiny.put(&key(2), InspTypedTree { n: 2 }).expect("put");
let tiny_page = tiny_db
.inspector()
.range(
InspTypedTree::NAME,
Bound::Unbounded,
Bound::Unbounded,
None,
limit2,
)
.expect("tiny page");
assert_eq!(page_keys(&tiny_page), vec![1, 2]);
assert!(tiny_page.next_key.is_none());
let err = inspector
.range(
name,
Bound::Included(&key_vec(2)),
Bound::Included(&key_vec(4)),
Some(&key_vec(9)),
limit2,
)
.unwrap_err();
assert!(matches!(err, InspectionError::InvalidBounds { .. }));
let terminal = inspector
.range(
name,
Bound::Included(&key_vec(2)),
Bound::Included(&key_vec(4)),
Some(&key_vec(4)),
limit2,
)
.expect("terminal after");
assert!(terminal.entries.is_empty());
assert!(terminal.next_key.is_none());
let mut_dir = tempdir().expect("mut tmp");
let mut_db = Db::open_test(mut_dir.path()).expect("mut open");
let mut_tree = mut_db
.open_typed_tree::<InspTypedTree, RapiraCodec, _>(asc_config(), NoHook, &[])
.expect("mut tree");
for id in 1..=5 {
mut_tree
.put(&key(id), InspTypedTree { n: id })
.expect("mut seed");
}
let mut_inspector = mut_db.inspector();
let page_a = mut_inspector
.range(
InspTypedTree::NAME,
Bound::Unbounded,
Bound::Unbounded,
None,
limit2,
)
.expect("before mutation");
mut_tree.delete(&key(1)).expect("delete between pages");
mut_tree
.put(&key(6), InspTypedTree { n: 6 })
.expect("insert between pages");
let page_b = mut_inspector
.range(
InspTypedTree::NAME,
Bound::Unbounded,
Bound::Unbounded,
page_a.next_key.as_deref(),
limit2,
)
.expect("after mutation");
assert!(!page_b.entries.is_empty());
let desc_name = InspTypedMapOrd::NAME;
let d1 = inspector
.range(desc_name, Bound::Unbounded, Bound::Unbounded, None, limit2)
.expect("desc p1");
assert_eq!(page_keys(&d1), vec![5, 4]);
assert_eq!(d1.next_key, Some(key_vec(4)));
let d2 = inspector
.range(
desc_name,
Bound::Unbounded,
Bound::Unbounded,
d1.next_key.as_deref(),
limit2,
)
.expect("desc p2");
assert_eq!(page_keys(&d2), vec![3, 2]);
assert_eq!(d2.next_key, Some(key_vec(2)));
let d3 = inspector
.range(
desc_name,
Bound::Unbounded,
Bound::Unbounded,
d2.next_key.as_deref(),
limit2,
)
.expect("desc p3");
assert_eq!(page_keys(&d3), vec![1]);
assert!(d3.next_key.is_none());
}
fn assert_unordered_snapshot_collection(
inspector: &armdb::armour::Inspector<'_>,
name: &str,
count: u64,
) {
let limit = NonZeroU32::new(10).unwrap();
let err = inspector
.range(name, Bound::Unbounded, Bound::Unbounded, None, limit)
.unwrap_err();
assert!(
matches!(
err,
InspectionError::UnsupportedOperation {
operation: InspectionOperation::Range,
..
}
),
"{name} range: {err:?}"
);
let err = inspector.prefix(name, &[], None, limit).unwrap_err();
assert!(
matches!(
err,
InspectionError::UnsupportedOperation {
operation: InspectionOperation::Prefix,
..
}
),
"{name} prefix: {err:?}"
);
let cap = NonZeroU32::new(count as u32).unwrap();
let entries = inspector
.snapshot_unordered(name, cap)
.expect("full snapshot");
assert_eq!(entries.len() as u64, count);
let keys: HashSet<_> = entries
.iter()
.map(|e| u64::from_be_bytes(e.key.as_slice().try_into().unwrap()))
.collect();
assert_eq!(keys, (1..=count).collect::<HashSet<_>>());
let err = inspector
.snapshot_unordered(name, NonZeroU32::new((count - 1) as u32).unwrap())
.unwrap_err();
assert!(
matches!(
err,
InspectionError::SnapshotTooLarge {
observed: Some(obs),
..
} if obs == count
),
"{name}: {err:?}"
);
}
#[test]
fn inspection_unordered_snapshot_matrix() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
let inspector = db.inspector();
assert_unordered_snapshot_collection(&inspector, InspTypedMapSnap::NAME, 3);
assert_unordered_snapshot_collection(&inspector, InspZeroMapSnapBc::NAME, 3);
assert_unordered_snapshot_collection(&inspector, InspZeroMapSnapFx::NAME, 3);
#[cfg(feature = "var-collections")]
assert_unordered_snapshot_collection(&inspector, InspVarMapSnap::NAME, 3);
let limit = NonZeroU32::new(1).unwrap();
let err = inspector
.snapshot_unordered(InspTypedTree::NAME, limit)
.unwrap_err();
assert!(matches!(
err,
InspectionError::UnsupportedOperation {
operation: InspectionOperation::SnapshotUnordered,
..
}
));
}
#[cfg(feature = "rpc")]
#[test]
fn inspection_rpc_wrong_length_key_semantics() {
let dir = tempdir().expect("tmp");
let db = open_matrix_db(dir.path(), false);
let rpc = db
.rpc_handler_for(InspTypedTree::NAME)
.expect("rpc handler");
let inspector = db.inspector();
let bad = &[1u8, 2, 3];
assert!(matches!(
RpcHandler::get(&*rpc, bad),
Err(DbError::KeyNotFound)
));
assert!(matches!(
inspector.get(InspTypedTree::NAME, bad),
Err(InspectionError::InvalidKey { .. })
));
}