use crate::index::{
config::{ByteVec, IndexType, Indexes, ValueMode, F32W, F64W},
serialization::{deserialize, serialize},
tree::{
compare, Index, KeyChanges, LeafEntry, Node, NodeRef, PageIter, PageIterBack, Value, ValueChange as TreeValue,
},
};
use crate::{
error::{PRes, PersyError},
persy::PersyImpl,
snapshot::SnapshotId,
transaction::Transaction,
PersyId,
};
use std::{
cmp::Ordering,
collections::{btree_map::Entry as BTreeEntry, hash_map::Entry, BTreeMap, HashMap},
iter::DoubleEndedIterator,
ops::{Bound, RangeBounds},
sync::Arc,
vec::IntoIter,
};
#[derive(Clone, Debug, PartialEq)]
pub enum ValueChange<V> {
ADD(V),
REMOVE(Option<V>),
}
#[derive(Clone)]
pub enum Change {
ADD(usize),
REMOVE(Option<usize>),
}
#[derive(Clone)]
pub struct Changes {
changes: Vec<Change>,
}
impl Changes {
fn new(change: Change) -> Changes {
Changes { changes: vec![change] }
}
fn push(&mut self, change: Change) {
self.changes.push(change);
}
}
macro_rules! impl_index_data_type {
($t:ty, $v:path, $v2:path, $t1:ty) => {
impl AddTo for $t {
fn add_to(self, vc: &mut ValueContainer) -> usize {
if let $v(ref mut v) = vc {
let l = v.len();
v.push(self.into());
l
} else {
panic!("wrong match from type and value container")
}
}
fn get_values(vc: &ValueContainer, changes: Changes) -> Vec<ValueChange<Self>> {
if let $v(ref v) = vc {
changes
.changes
.iter()
.map(|c| match c {
Change::ADD(p) => ValueChange::ADD(v[*p].clone().into()),
Change::REMOVE(o) => ValueChange::REMOVE(o.map(|p| v[p].clone().into())),
})
.collect()
} else {
panic!("wrong match from type and value container")
}
}
fn add_to_entries(self, vc: &mut EntriesContainer, change: Change) {
if let $v2(ref mut v) = vc {
match v.entry(self.into()) {
BTreeEntry::Occupied(ref mut o) => {
o.get_mut().push(change);
}
BTreeEntry::Vacant(va) => {
va.insert(Changes::new(change));
}
}
} else {
panic!("wrong match from type and value container")
}
}
fn get_changes(&self, vc: &EntriesContainer) -> Option<Changes> {
if let $v2(ref v) = vc {
let k: $t1 = self.clone().into();
v.get(&k).map(Clone::clone)
} else {
None
}
}
fn keys(vc: &EntriesContainer) -> IntoIter<Self> {
if let $v2(v) = vc {
v.keys()
.map(|x| x.clone().into())
.collect::<Vec<Self>>()
.into_iter()
} else {
panic!("wrong match from type and value container")
}
}
fn range<R>(vc: &EntriesContainer, range: R) -> IntoIter<Self>
where
R: RangeBounds<Self>,
{
if let $v2(v) = vc {
let into_range: (Bound<$t1>, Bound<$t1>) =
(map_bound(range.start_bound()), map_bound(range.end_bound()));
v.range(into_range)
.map(|x| x.0.clone().into())
.collect::<Vec<Self>>()
.into_iter()
} else {
panic!("wrong match from type and value container")
}
}
fn new_entries() -> EntriesContainer {
$v2(BTreeMap::new())
}
fn new_values() -> ValueContainer {
$v(Vec::new())
}
}
};
}
macro_rules! container_enums {
($($variant:ident<$t:ty,$mt:ty>),+,) => {
#[derive(Clone)]
pub enum EntriesContainer {
$(
$variant(BTreeMap<$mt, Changes>),
)+
}
#[derive(Clone)]
pub enum ValueContainer {
$(
$variant(Vec<$mt>),
)+
}
fn eapplier(
keys: &EntriesContainer,
values: &ValueContainer,
index_name: &str,
persy: &PersyImpl,
tx: &mut Transaction,
) -> PRes<()> {
match keys {
$(
EntriesContainer::$variant(k) => valapplier::<$t, $mt>(values, k, index_name, persy, tx),
)+
}
}
fn valapplier<K, MK>(
values: &ValueContainer,
k: &BTreeMap<MK, Changes>,
index_name: &str,
persy: &PersyImpl,
tx: &mut Transaction,
) -> PRes<()>
where
K: IndexType,
MK: Into<K> + Clone,
{
match values {
$(
ValueContainer::$variant(v) => apply_to_index::<K, $t, MK, $mt>(persy, tx, index_name, k, v),
)+
}
}
$(
impl_index_data_type!($t, ValueContainer::$variant, EntriesContainer::$variant, $mt);
)+
}
}
container_enums!(
U8<u8, u8>,
U16<u16, u16>,
U32<u32, u32>,
U64<u64, u64>,
U128<u128, u128>,
I8<i8, i8>,
I16<i16, i16>,
I32<i32, i32>,
I64<i64, i64>,
I128<i128, i128>,
F32W<f32, F32W>,
F64W<f64, F64W>,
STRING<String, String>,
PERSYID<PersyId, PersyId>,
BYTEVEC<ByteVec, ByteVec>,
);
pub trait AddTo: Sized {
fn add_to(self, vc: &mut ValueContainer) -> usize;
fn get_values(vc: &ValueContainer, changes: Changes) -> Vec<ValueChange<Self>>;
fn add_to_entries(self, vc: &mut EntriesContainer, change: Change);
fn get_changes(&self, vc: &EntriesContainer) -> Option<Changes>;
fn new_entries() -> EntriesContainer;
fn new_values() -> ValueContainer;
fn keys(vc: &EntriesContainer) -> IntoIter<Self>;
fn range<R>(vc: &EntriesContainer, range: R) -> IntoIter<Self>
where
R: RangeBounds<Self>;
}
fn apply_to_index<K, V, MK, MV>(
persy: &PersyImpl,
tx: &mut Transaction,
index_name: &str,
keys: &BTreeMap<MK, Changes>,
values: &[MV],
) -> PRes<()>
where
K: IndexType,
V: IndexType,
MK: Into<K> + Clone,
MV: Into<V> + Clone,
{
let mut changes = Vec::new();
for (k, c) in keys.iter() {
let mut vals = Vec::new();
for ch in &c.changes {
vals.push(match *ch {
Change::ADD(pos) => TreeValue::ADD(values[pos].clone().into()),
Change::REMOVE(pos) => TreeValue::REMOVE(pos.map(|p| values[p].clone().into())),
});
}
changes.push(KeyChanges::new(k.clone().into(), vals));
}
let root;
{
let mut index = Indexes::check_and_get_index_keeper::<K, V>(persy, Some(tx), None, index_name)?;
index.apply(&changes)?;
index.update_changed()?;
root = IndexKeeper::<K, V>::get_root(&index)?;
}
Indexes::update_index_root(persy, tx, index_name, root)?;
Ok(())
}
fn map_bound<T: Into<T1> + Clone, T1>(b: Bound<&T>) -> Bound<T1> {
match b {
Bound::Excluded(x) => Bound::Excluded((*x).clone().into()),
Bound::Included(x) => Bound::Included((*x).clone().into()),
Bound::Unbounded => Bound::Unbounded,
}
}
pub struct IndexTransactionKeeper {
indexex_changes: HashMap<String, (EntriesContainer, ValueContainer)>,
}
impl IndexTransactionKeeper {
pub fn new() -> IndexTransactionKeeper {
IndexTransactionKeeper {
indexex_changes: HashMap::new(),
}
}
pub fn put<K, V>(&mut self, index: &str, k: K, v: V)
where
K: IndexType,
V: IndexType,
{
match self.indexex_changes.entry(index.to_string()) {
Entry::Occupied(ref mut o) => {
let pos = v.add_to(&mut o.get_mut().1);
k.add_to_entries(&mut o.get_mut().0, Change::ADD(pos));
}
Entry::Vacant(va) => {
let mut values = V::new_values();
let mut keys = K::new_entries();
let pos = v.add_to(&mut values);
k.add_to_entries(&mut keys, Change::ADD(pos));
va.insert((keys, values));
}
}
}
pub fn remove<K, V>(&mut self, index: &str, k: K, v: Option<V>)
where
K: IndexType,
V: IndexType,
{
match self.indexex_changes.entry(index.to_string()) {
Entry::Occupied(ref mut o) => {
let pos = if let Some(val) = v {
Some(val.add_to(&mut o.get_mut().1))
} else {
None
};
k.add_to_entries(&mut o.get_mut().0, Change::REMOVE(pos));
}
Entry::Vacant(va) => {
let mut values = V::new_values();
let mut keys = K::new_entries();
let pos = if let Some(val) = v {
Some(val.add_to(&mut values))
} else {
None
};
k.add_to_entries(&mut keys, Change::REMOVE(pos));
va.insert((keys, values));
}
}
}
pub fn get_changes<K, V>(&self, index: &str, k: &K) -> Option<Vec<ValueChange<V>>>
where
K: IndexType,
V: IndexType,
{
self.indexex_changes
.get(index)
.map(|ref o| k.get_changes(&o.0).map(|c| V::get_values(&o.1, c)))
.and_then(std::convert::identity)
}
pub fn apply_changes<K, V>(
&self,
index_name: &str,
vm: ValueMode,
k: &K,
pers: Option<Value<V>>,
) -> PRes<Option<Value<V>>>
where
K: IndexType,
V: IndexType,
{
let mut result = pers;
if let Some(key_changes) = self.get_changes::<K, V>(index_name, k) {
for change in key_changes {
result = match change {
ValueChange::ADD(add_value) => Some(if let Some(s_result) = result {
match s_result {
Value::SINGLE(v) => match vm {
ValueMode::REPLACE => Value::SINGLE(add_value),
ValueMode::EXCLUSIVE => {
if compare(&v, &add_value) == Ordering::Equal {
Value::SINGLE(v)
} else {
return Err(PersyError::IndexDuplicateKey(
index_name.to_string(),
format!("{}", k),
));
}
}
ValueMode::CLUSTER => {
if compare(&v, &add_value) == Ordering::Equal {
Value::SINGLE(v)
} else {
Value::CLUSTER(vec![v, add_value])
}
}
},
Value::CLUSTER(mut values) => {
if let Ok(pos) = values.binary_search_by(|x| compare(x, &add_value)) {
values.insert(pos, add_value);
}
Value::CLUSTER(values)
}
}
} else {
Value::SINGLE(add_value)
}),
ValueChange::REMOVE(rv) => rv.and_then(|remove_value| {
result.and_then(|s_result| match s_result {
Value::SINGLE(v) => {
if compare(&v, &remove_value) == Ordering::Equal {
None
} else {
Some(Value::SINGLE(v))
}
}
Value::CLUSTER(mut values) => {
if let Ok(pos) = values.binary_search_by(|x| compare(x, &remove_value)) {
values.remove(pos);
}
Some(if values.len() == 1 {
Value::SINGLE(values.pop().unwrap())
} else {
Value::CLUSTER(values)
})
}
})
}),
};
}
}
Ok(result)
}
pub fn apply(&self, persy: &PersyImpl, tx: &mut Transaction) -> PRes<()> {
for (index, values) in &self.indexex_changes {
eapplier(&values.0, &values.1, index, persy, tx)?;
}
Ok(())
}
pub fn range<K, V, R>(&self, index_name: &str, range: R) -> Option<IntoIter<K>>
where
K: IndexType,
V: IndexType,
R: RangeBounds<K>,
{
self.indexex_changes.get(index_name).map(|x| K::range(&x.0, range))
}
pub fn changed_indexes(&self) -> Vec<String> {
self.indexex_changes.keys().cloned().collect()
}
}
pub trait IndexKeeper<K, V> {
fn load(&mut self, node: &NodeRef) -> PRes<Node<K, V>>;
fn insert(&mut self, node: Node<K, V>) -> PRes<NodeRef>;
fn update(&mut self, node_ref: &NodeRef, node: Node<K, V>) -> PRes<()>;
fn delete(&mut self, node: &NodeRef) -> PRes<()>;
fn get_root(&self) -> PRes<Option<NodeRef>>;
fn set_root(&mut self, root: Option<NodeRef>) -> PRes<()>;
fn bottom_limit(&self) -> usize;
fn top_limit(&self) -> usize;
fn value_mode(&self) -> ValueMode;
fn index_name(&self) -> &String;
}
pub struct IndexSegmentKeeper<'a, K: IndexType, V: IndexType> {
name: String,
segment: String,
root: Option<NodeRef>,
store: &'a PersyImpl,
tx: Option<&'a mut Transaction>,
snapshot: Option<SnapshotId>,
value_mode: ValueMode,
changed: Option<HashMap<NodeRef, Node<K, V>>>,
}
impl<'a, K: IndexType, V: IndexType> IndexSegmentKeeper<'a, K, V> {
pub fn new(
name: &str,
segment: &str,
root: Option<NodeRef>,
store: &'a PersyImpl,
tx: Option<&'a mut Transaction>,
snapshot: Option<SnapshotId>,
value_mode: ValueMode,
) -> IndexSegmentKeeper<'a, K, V> {
IndexSegmentKeeper {
name: name.to_string(),
segment: segment.to_string(),
root,
store,
tx,
snapshot,
value_mode,
changed: None,
}
}
pub fn update_changed(&mut self) -> PRes<()> {
if let Some(ref mut tx) = self.tx {
if let Some(m) = &self.changed {
for (node_ref, node) in m {
self.store
.update_record(tx, &self.segment, &node_ref, &serialize(node.clone())?)?;
}
}
}
Ok(())
}
}
impl<'a, K: IndexType, V: IndexType> IndexKeeper<K, V> for IndexSegmentKeeper<'a, K, V> {
fn load(&mut self, node: &NodeRef) -> PRes<Node<K, V>> {
if let Some(m) = &mut self.changed {
if let Some(n) = m.get(node) {
return Ok(n.clone());
}
}
deserialize(
if let Some(ref mut tx) = self.tx {
self.store.read_record_tx(tx, &self.segment, &node)
} else if let Some(snapshot_id) = self.snapshot {
self.store.read_record_snapshot(&self.segment, &node, snapshot_id)
} else {
self.store.read_record(&self.segment, &node)
}?
.as_ref()
.unwrap(),
)
}
fn insert(&mut self, node: Node<K, V>) -> PRes<NodeRef> {
if let Some(ref mut tx) = self.tx {
let node_ref = self.store.insert_record(tx, &self.segment, &serialize(node.clone())?)?;
self.changed
.get_or_insert(HashMap::new())
.insert(node_ref.clone(), node);
Ok(node_ref)
} else {
panic!("should never get here");
}
}
fn update(&mut self, node_ref: &NodeRef, node: Node<K, V>) -> PRes<()> {
self.changed
.get_or_insert(HashMap::new())
.insert(node_ref.clone(), node);
Ok(())
}
fn delete(&mut self, node: &NodeRef) -> PRes<()> {
if let Some(m) = &mut self.changed {
m.remove(node);
}
if let Some(ref mut tx) = self.tx {
self.store.delete_record(tx, &self.segment, &node)?;
Ok(())
} else {
panic!("should not get here");
}
}
fn get_root(&self) -> PRes<Option<NodeRef>> {
Ok(self.root.clone())
}
fn set_root(&mut self, root: Option<NodeRef>) -> PRes<()> {
self.root = root;
Ok(())
}
fn bottom_limit(&self) -> usize {
10
}
fn top_limit(&self) -> usize {
30
}
fn value_mode(&self) -> ValueMode {
self.value_mode.clone()
}
fn index_name(&self) -> &String {
&self.name
}
}
pub struct TxIndexRawIter<K: IndexType, V: IndexType> {
index_name: String,
in_tx: Option<IntoIter<K>>,
persistent: Option<IndexRawIter<K, V>>,
in_tx_front: Option<Option<K>>,
persistent_front: Option<Option<(K, Value<V>)>>,
in_tx_back: Option<Option<K>>,
persistent_back: Option<Option<(K, Value<V>)>>,
value_mode: ValueMode,
}
impl<K, V> TxIndexRawIter<K, V>
where
K: IndexType,
V: IndexType,
{
pub fn new(
index_name: &str,
in_tx: Option<IntoIter<K>>,
persistent: Option<IndexRawIter<K, V>>,
value_mode: ValueMode,
) -> TxIndexRawIter<K, V> {
TxIndexRawIter {
index_name: index_name.to_owned(),
in_tx,
persistent,
in_tx_front: None,
persistent_front: None,
in_tx_back: None,
persistent_back: None,
value_mode,
}
}
fn apply_changes(
tx: &mut Transaction,
vm: ValueMode,
index: &str,
k: K,
pers: Option<Value<V>>,
) -> Option<(K, Value<V>)> {
tx.apply_changes(vm, index, &k, pers).unwrap_or(None).map(|v| (k, v))
}
pub fn next(&mut self, persy_impl: &Arc<PersyImpl>, tx: &mut Transaction) -> Option<(K, Value<V>)> {
let vm = self.value_mode.clone();
let index = &self.index_name;
let apply_changes = |k, o| Self::apply_changes(tx, vm, index, k, o);
match (&mut self.in_tx, &mut self.persistent) {
(Some(it), Some(pers)) => {
match (
self.in_tx_front.get_or_insert_with(|| it.next()).clone(),
self.persistent_front
.get_or_insert_with(|| pers.next(persy_impl))
.clone(),
) {
(Some(tx_k), Some((pers_k, vals))) => match tx_k.cmp(&pers_k) {
Ordering::Less => {
self.in_tx_front = None;
apply_changes(tx_k, None)
}
Ordering::Equal => {
self.in_tx_front = None;
self.persistent_front = None;
apply_changes(tx_k, Some(vals))
}
Ordering::Greater => {
self.persistent_front = None;
Some((pers_k, vals))
}
},
(Some(tx_k), None) => {
self.in_tx_front = None;
apply_changes(tx_k, None)
}
(None, Some((pers_k, vals))) => {
self.persistent_front = None;
Some((pers_k, vals))
}
(None, None) => None,
}
}
(Some(it), None) => apply_changes(it.next().unwrap(), None),
(None, Some(pers)) => pers.next(persy_impl),
(None, None) => None,
}
}
pub fn next_back(&mut self, persy_impl: &Arc<PersyImpl>, tx: &mut Transaction) -> Option<(K, Value<V>)> {
let vm = self.value_mode.clone();
let index = &self.index_name;
let apply_changes = |k, o| Self::apply_changes(tx, vm, index, k, o);
match (&mut self.in_tx, &mut self.persistent) {
(Some(it), Some(pers)) => {
match (
self.in_tx_back.get_or_insert_with(|| it.next_back()).clone(),
self.persistent_back
.get_or_insert_with(|| pers.next_back(persy_impl))
.clone(),
) {
(Some(tx_k), Some((pers_k, vals))) => match tx_k.cmp(&pers_k) {
Ordering::Less => {
self.persistent_back = None;
Some((pers_k, vals))
}
Ordering::Equal => {
self.in_tx_back = None;
self.persistent_back = None;
apply_changes(tx_k, Some(vals))
}
Ordering::Greater => {
self.in_tx_back = None;
apply_changes(tx_k, None)
}
},
(Some(tx_k), None) => {
self.in_tx_back = None;
apply_changes(tx_k, None)
}
(None, Some((pers_k, vals))) => {
self.persistent_back = None;
Some((pers_k, vals))
}
(None, None) => None,
}
}
(Some(it), None) => apply_changes(it.next_back().unwrap(), None),
(None, Some(pers)) => pers.next_back(persy_impl),
(None, None) => None,
}
}
}
pub struct IndexRawIter<K: IndexType, V: IndexType> {
index_name: String,
read_snapshot: SnapshotId,
iter: PageIter<K, V>,
back: PageIterBack<K, V>,
}
impl<K, V> IndexRawIter<K, V>
where
K: IndexType,
V: IndexType,
{
pub fn new(
index_name: &str,
read_snapshot: SnapshotId,
iter: PageIter<K, V>,
back: PageIterBack<K, V>,
) -> IndexRawIter<K, V> {
IndexRawIter {
index_name: index_name.to_owned(),
read_snapshot,
iter,
back,
}
}
pub fn next(&mut self, persy_impl: &Arc<PersyImpl>) -> Option<(K, Value<V>)> {
loop {
let back_keep = self.back_peek(persy_impl);
if let (Some(s), Some(e)) = (self.iter.iter.peek(), back_keep) {
if s.key.cmp(&e.key) == Ordering::Greater {
break None;
}
}
if let Some(n) = self.iter.iter.next() {
break Some((n.key.clone(), n.value.clone()));
} else if let Some(next_id) = self.iter.next.clone() {
if let Ok(iter) = persy_impl.index_next(&self.index_name, self.read_snapshot, &next_id) {
self.iter = iter;
continue;
}
}
break None;
}
}
fn back_peek(&mut self, persy_impl: &Arc<PersyImpl>) -> Option<LeafEntry<K, V>> {
loop {
if let Some(peek_keep) = self.back.iter.peek().cloned() {
break Some(peek_keep);
} else if let Some(back_id) = &self.back.prev {
if let Ok(back) = persy_impl.index_next_back(&self.index_name, self.read_snapshot, &back_id) {
self.back = back;
continue;
}
}
break None;
}
}
fn front_peek(&mut self, persy_impl: &Arc<PersyImpl>) -> Option<LeafEntry<K, V>> {
loop {
if let Some(peek_keep) = self.iter.iter.peek().cloned() {
break Some(peek_keep);
} else if let Some(next_id) = &self.iter.next {
if let Ok(iter) = persy_impl.index_next(&self.index_name, self.read_snapshot, &next_id) {
self.iter = iter;
continue;
}
}
break None;
}
}
pub fn next_back(&mut self, persy_impl: &Arc<PersyImpl>) -> Option<(K, Value<V>)> {
loop {
let front_keep = self.front_peek(persy_impl);
if let (Some(s), Some(e)) = (self.back.iter.peek(), front_keep) {
if s.key.cmp(&e.key) == Ordering::Less {
break None;
}
}
if let Some(n) = self.back.iter.next() {
break Some((n.key.clone(), n.value.clone()));
} else if let Some(prev_id) = self.back.prev.clone() {
if let Ok(back) = persy_impl.index_next_back(&self.index_name, self.read_snapshot, &prev_id) {
self.back = back;
continue;
}
}
break None;
}
}
pub fn release(&self, persy_impl: &Arc<PersyImpl>) -> PRes<()> {
persy_impl.release(self.read_snapshot)
}
}
#[cfg(test)]
mod tests {
use super::{ByteVec, IndexTransactionKeeper, IndexType, ValueChange};
use crate::id::{PersyId, RecRef};
use std::fmt::Debug;
fn keeper_test_for_type<K: IndexType + PartialEq, V: IndexType + Debug + PartialEq>(k: K, dk: K, v: V) {
let name = "index";
let mut keeper = IndexTransactionKeeper::new();
keeper.put(name, k.clone(), v.clone());
let ret = keeper.get_changes(name, &k);
assert_eq!(ret, Some(vec![ValueChange::ADD(v.clone())]));
keeper.remove(name, dk.clone(), Some(v.clone()));
let ret = keeper.get_changes(name, &dk);
assert_eq!(ret, Some(vec![ValueChange::REMOVE(Some(v))]));
}
#[test]
fn simple_tx_keeper_test() {
keeper_test_for_type::<u8, u8>(10, 15, 10);
keeper_test_for_type::<u16, u16>(10, 15, 10);
keeper_test_for_type::<u32, u32>(10, 15, 10);
keeper_test_for_type::<u64, u64>(10, 15, 10);
keeper_test_for_type::<u128, u128>(10, 15, 10);
keeper_test_for_type::<i8, i8>(10, 15, 10);
keeper_test_for_type::<i16, i16>(10, 15, 10);
keeper_test_for_type::<i32, i32>(10, 15, 10);
keeper_test_for_type::<i64, i64>(10, 15, 10);
keeper_test_for_type::<i128, i128>(10, 15, 10);
keeper_test_for_type::<f32, f32>(20.0, 10.0, 20.0);
keeper_test_for_type::<f64, f64>(20.0, 10.0, 20.0);
keeper_test_for_type::<String, String>("a".to_string(), "b".to_string(), "a".to_string());
keeper_test_for_type::<ByteVec, ByteVec>(vec![0, 1].into(), vec![0, 2].into(), vec![0, 1].into());
let id = PersyId(RecRef::new(10, 20));
let id1 = PersyId(RecRef::new(20, 20));
let id2 = PersyId(RecRef::new(30, 20));
keeper_test_for_type::<PersyId, PersyId>(id, id1, id2);
}
}