use super::*;
use crate::storage::warp_table::WarpTable;
use alloc::vec::Vec;
#[derive(Clone, Debug, Eq, PartialEq, WarpTableRow)]
#[index(primary=u32)]
struct Keys {
#[index(secondary, range)]
room: u32,
#[index(unique)]
code: u32,
}
impl PersistentRow for Keys {
fn own(row: KeysRef<'_>) -> Self {
Self {
room: *row.room,
code: *row.code,
}
}
fn retained_bytes(&self) -> usize {
8
}
}
fn located(room: u32, code: u32, offset: u64) -> Located<Keys> {
Located {
keys: Keys { room, code },
location: RecordLocation {
generation: 1,
offset,
bytes: 64,
},
}
}
#[test]
fn index_membership_and_atomic_unique_swaps_do_not_require_record_payloads() -> Result<(), Error> {
let mut limits = limits();
limits.row_bytes = 64;
let mut authority = Authority::<Located<Keys>>::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, located(3, 10, 0)), (2, located(4, 20, 64))],
outcomes: vec![],
},
1024,
limits,
Replication::Disabled,
)?;
let changes = vec![
Change::Replace {
id: 1,
row: located(4, 20, 128),
},
Change::Replace {
id: 2,
row: located(3, 10, 192),
},
];
authority.validate_changes(&changes)?;
let inspect =
|authority: &Authority<Located<Keys>>, guarantee: Guarantee| -> Result<Vec<u32>, Error> {
match authority.read(ReadRequest { guarantee }) {
ReadOutcome::Ready { table, .. } => Ok(IndexView::new(table)
.select(|query| query.find_room(3))
.ids()
.collect()),
ReadOutcome::Pending => Err(Error::Busy),
ReadOutcome::Refused(error) => Err(error),
}
};
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [1]);
let session = authority.open_session()?;
authority
.step(Submit {
receipt: None,
operation: authority.operation(session, 1),
expected: Revision(0),
guarantee: Guarantee::Accepted,
changes,
})
.map_err(|(_request, error)| error)?;
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [2]);
assert_eq!(inspect(&authority, Guarantee::LocallyDurable)?, [1]);
let attempt = authority
.prepare(Guarantee::LocallyDurable)?
.ok_or(Error::Busy)?;
authority.step(Complete {
attempt,
result: PhysicalResult::Durable,
})?;
assert_eq!(inspect(&authority, Guarantee::LocallyDurable)?, [2]);
assert_eq!(
authority.validate_changes(&[Change::Replace {
id: 1,
row: located(3, 10, 256)
}]),
Err(Error::UniqueConflict)
);
assert_eq!(inspect(&authority, Guarantee::Accepted)?, [2]);
Ok(())
}
#[test]
fn bounded_target_and_change_sets_remember_a_discarded_refusal() -> Result<(), Error> {
let mut targets = Targets::new(2)?;
targets.push(1)?;
targets.push(1)?;
targets.push(2)?;
assert_eq!(targets.push(3), Err(Error::Full));
targets.push(1)?;
assert_eq!(targets.finish(), Err(Error::Full));
let mut changes = ChangeSet::<Keys>::new(1)?;
assert!(changes.push(Change::Remove { id: 1 }).is_ok());
assert!(changes.push(Change::Remove { id: 2 }).is_err());
assert!(matches!(changes.finish(), Err(Error::Full)));
assert!(matches!(
ChangeSet::<Keys>::new(0),
Err(Error::InvalidBounds)
));
assert!(matches!(Targets::<u32>::new(0), Err(Error::InvalidBounds)));
Ok(())
}
#[test]
fn located_columns_keep_keys_locations_and_indexes_aligned() -> Result<(), Error> {
let mut table =
WarpTable::<Located<Keys>>::try_new(limits().rows).map_err(|_error| Error::Allocation)?;
assert!(table.query().is_empty());
assert_eq!(table.query().len(), 0);
assert!(table.query().get(1).is_none());
super::super::changes::insert(&mut table, 1, located(3, 10, 0))?;
super::super::changes::insert(&mut table, 2, located(4, 20, 64))?;
assert!(!table.query().is_empty());
assert_eq!(table.query().len(), 2);
assert_eq!(table.query().ids(), [1, 2]);
assert_eq!(*table.query().get(1).ok_or(Error::Absent)?.code, 10);
assert!(matches!(
table.replace(1, located(4, 20, 128)),
crate::storage::warp_table::WarpTableReplaceOutcome::UniqueConflict { .. }
));
assert!(matches!(
table.replace(1, located(9, 30, 128)),
crate::storage::warp_table::WarpTableReplaceOutcome::Replaced { .. }
));
assert!(
table
.query()
.select(|query| query.find_room(3))
.ids()
.next()
.is_none()
);
assert_eq!(
table
.query()
.select(|query| query.find_room(9))
.ids()
.collect::<Vec<_>>(),
[1]
);
match table.remove(1) {
crate::storage::warp_table::WarpTableRemoveOutcome::Removed { removed, .. } => {
assert_eq!(removed.location.offset, 128)
}
crate::storage::warp_table::WarpTableRemoveOutcome::Absent { .. } => {
return Err(Error::Absent);
}
}
match table.get(2) {
WarpTableGetOutcome::Present { row } => {
assert_eq!(*row.keys.room, 4);
assert_eq!(row.location.offset, 64);
}
WarpTableGetOutcome::Absent { .. } => return Err(Error::Absent),
}
let ids: Vec<_> = table
.query()
.select(|query| query.find_room(4))
.ids()
.collect();
assert_eq!(ids, [2]);
Ok(())
}
#[test]
fn resident_byte_admission_counts_record_locations_as_well_as_keys() {
let mut limits = limits();
limits.row_bytes = 8;
assert!(matches!(
Authority::<Located<Keys>>::recover(
Recovery {
history: HistoryId([1; 16]),
revision: Revision(0),
reserved_through: 0,
rows: vec![(1, located(3, 10, 0))],
outcomes: vec![]
},
1024,
limits,
Replication::Disabled
),
Err(Error::InvalidBounds)
));
}
#[test]
fn located_mutation_changes_the_location_without_losing_indexed_keys() -> Result<(), Error> {
use crate::storage::warp_table::{WarpTableRowGetOutcome, WarpTableUpdateOutcome};
let mut table =
WarpTable::<Located<Keys>>::try_new(limits().rows).map_err(|_| Error::Allocation)?;
super::super::changes::insert(&mut table, 1, located(3, 10, 0))?;
let () = table.query_mut();
let () = table.columns_mut();
assert_eq!(
table.update(1, |row| {
row.location.offset = 128;
(*row.keys.room(), *row.keys.code())
}),
WarpTableUpdateOutcome::Updated {
id: 1,
row: 0,
output: (3, 10)
}
);
let WarpTableGetOutcome::Present { row } = table.get(1) else {
return Err(Error::Absent);
};
assert_eq!(row.location.offset, 128);
assert_eq!(*row.keys.code, 10);
assert_eq!(
IndexView::new(&table)
.select(|query| query.find_room(3))
.ids()
.collect::<Vec<_>>(),
[1]
);
let mut columns = Located::<Keys>::try_new_columns(1).map_err(|_| Error::Allocation)?;
let mut indexes = Located::<Keys>::try_new_indexes(1).map_err(|_| Error::Allocation)?;
assert!(matches!(
Located::<Keys>::get_mut(&1, &mut columns, &mut indexes, 0),
Err(
crate::storage::warp_table::WarpTableRowGetError::ColumnUnavailable {
column: 0,
row: 0
}
)
));
Located::<Keys>::append(&mut columns, located(3, 10, 0));
let WarpTableRowGetOutcome::Present { row } =
Located::<Keys>::get_mut(&1, &mut columns, &mut indexes, 0).map_err(|_| Error::Corrupt)?;
row.location.generation = 2;
let WarpTableRowGetOutcome::Present { row } =
Located::<Keys>::get(&1, &columns, 0).map_err(|_| Error::Corrupt)?;
assert_eq!(row.location.generation, 2);
assert!(matches!(
Located::<Keys>::replace(&mut columns, 1, located(9, 99, 9)),
Err(
crate::storage::warp_table::WarpTableRowReplaceError::ColumnUnavailable {
column: 0,
row: 1,
rejected: Located {
keys: Keys { room: 9, code: 99 },
location: RecordLocation { offset: 9, .. }
}
}
)
));
let mut targets = Targets::new(2)?;
targets.push(1)?;
targets.push(1)?;
targets.push(2)?;
assert_eq!(targets.finish()?, [1, 2]);
Ok(())
}
#[test]
fn located_detached_queries_delegate_the_key_indexes() -> Result<(), Error> {
let mut table =
WarpTable::<Located<Keys>>::try_new(limits().rows).map_err(|_| Error::Allocation)?;
super::super::changes::insert(&mut table, 1, located(3, 10, 0))?;
let selection = Located::<Keys>::detached_query(table.__indexes()).find_room(3);
let selection =
crate::storage::warp_table::query::bind_detached_selection(&table, selection.into());
assert_eq!(selection.ids().collect::<Vec<_>>(), [1]);
let selection = Located::<Keys>::detached_query(table.__indexes())
.find_room_in(99..=100)
.map_err(|_| Error::Corrupt)?;
let selection =
crate::storage::warp_table::query::bind_detached_selection(&table, selection.into());
assert!(selection.ids().next().is_none());
Ok(())
}