use crate::bits::Bitset;
use crate::idxsets::intersection::Intersection;
use crate::internal::mr::rvec::RVec;
use crate::internal::mr::summarize::{Summarize, SummaryRules};
use crate::traits::idxset::IdxSet;
use crate::traits::query::Query;
use crate::traits::record::Record;
use crate::traits::valid_key::{BorrowedKey, ValidKey};
use crate::types::chunk_storage::ChunkStorage;
use crate::types::storage::Storage;
use std::borrow::Borrow;
use std::borrow::Cow;
use std::collections::HashMap;
use std::collections::{BTreeSet, HashSet};
use std::fmt::Debug;
use std::iter::Map;
use std::sync::Arc;
use std::sync::RwLock;
pub struct MatchingSecondaryIndex<'a, Q, ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey + Borrow<IndexKey>,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
query: Q,
secondary_index: SecondaryIndex<ChunkKey, Element, IndexKeys, IndexKey>,
index_key: Cow<'a, IndexKey>,
}
impl<'a, Q, ChunkKey, Element, IndexKeys, IndexKey> Clone
for MatchingSecondaryIndex<'a, Q, ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey + Borrow<IndexKey>,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
Q: Clone,
{
fn clone(&self) -> Self {
MatchingSecondaryIndex {
query: self.query.clone(),
secondary_index: self.secondary_index.clone(),
index_key: self.index_key.clone(),
}
}
}
struct ChunkSecondaryIndex<IndexKey>
where
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
{
reverse_index: HashMap<IndexKey::Owned, Bitset>,
}
pub struct SecondaryIndex<ChunkKey, Element, IndexKeys, IndexKey>(
Arc<RwLock<SecondaryIndexImpl<ChunkKey, Element, IndexKeys, IndexKey>>>,
)
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>;
impl<ChunkKey, Element, IndexKeys, IndexKey> Clone
for SecondaryIndex<ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
fn clone(&self) -> Self {
SecondaryIndex(Arc::clone(&self.0))
}
}
struct SecondaryIndexImpl<ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
parent_id: u64,
gc_chunk_list: RVec<Option<ChunkKey::Owned>>,
rules: Arc<SummaryRules<Element, IndexKeys, ChunkSecondaryIndex<IndexKey>>>,
index: HashMap<
ChunkKey::Owned,
Summarize<Element, IndexKeys, ChunkSecondaryIndex<IndexKey>>,
crate::internal::hasher::HasherImpl,
>,
}
impl<ChunkKey, Element, IndexKeys, IndexKey> SecondaryIndex<ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
pub fn new<ItemKey, F>(storage: &Storage<ChunkKey, ItemKey, Element>, f: F) -> Self
where
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
F: Fn(&Element) -> Cow<IndexKeys> + Clone + Send + Sync + 'static,
{
SecondaryIndex(Arc::new(RwLock::new(SecondaryIndexImpl {
parent_id: storage.id(),
gc_chunk_list: RVec::default(),
index: HashMap::with_hasher(crate::internal::hasher::HasherImpl::default()),
rules: Arc::new(
SecondaryIndexImpl::<ChunkKey, Element, IndexKeys, IndexKey>::indexing_rules(f),
),
})))
}
pub fn validate<ItemKey>(&self, parent: &Storage<ChunkKey, ItemKey, Element>)
where
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
{
self.0.write().unwrap().validate(parent);
}
}
impl<ChunkKey, Element, IndexKeys, IndexKey>
SecondaryIndexImpl<ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
fn indexing_rules<F>(f: F) -> SummaryRules<Element, IndexKeys, ChunkSecondaryIndex<IndexKey>>
where
F: Fn(&Element) -> Cow<IndexKeys> + Clone + Send + Sync + 'static,
{
SummaryRules {
map: Arc::new(move |element, old_index_keys, _internal_idx| {
let new_index_keys = f(element);
if old_index_keys != new_index_keys.borrow() {
Some(new_index_keys.into_owned())
} else {
None
}
}),
contribute: Arc::new(|new_index_keys, internal_idx, summary| {
for new_index_key in new_index_keys.iter_keys() {
let idx_set = summary
.reverse_index
.entry(new_index_key.into_owned())
.or_insert_with(Bitset::default);
idx_set.set(internal_idx);
}
}),
uncontribute: Arc::new(|old_index_keys, internal_idx, summary| {
for old_index_key in old_index_keys.iter_keys() {
let mut remove = false;
if let Some(idx_set) = summary.reverse_index.get_mut(old_index_key.borrow()) {
idx_set.unset(internal_idx);
if idx_set.is_empty() {
remove = true;
}
}
if remove {
summary.reverse_index.remove(old_index_key.borrow());
}
}
}),
}
}
pub(crate) fn update_chunk<ItemKey>(
&mut self,
chunk_key: &ChunkKey,
chunk_storage: &ChunkStorage<ChunkKey, ItemKey, Element>,
) where
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
{
let index = &mut self.index;
let rules = &self.rules;
let internal_storage = chunk_storage.internal_rvec();
index
.entry(chunk_key.to_owned())
.or_insert_with(|| Summarize::new(&internal_storage, Arc::clone(rules)))
.update(&internal_storage);
}
pub(crate) fn gc<ItemKey>(&mut self, parent: &Storage<ChunkKey, ItemKey, Element>)
where
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
{
parent.gc(&mut self.gc_chunk_list, &mut self.index);
}
pub fn validate<ItemKey>(&mut self, parent: &Storage<ChunkKey, ItemKey, Element>)
where
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
{
self.gc(parent);
for chunk_key in self.index.keys() {
assert!(parent.internal_idx_of(chunk_key.borrow()).is_some());
}
}
}
impl<'a, Q, ChunkKey, Element, IndexKeys, IndexKey>
MatchingSecondaryIndex<'a, Q, ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey + Borrow<IndexKey>,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
{
pub(crate) fn new(
query: Q,
secondary_index: &SecondaryIndex<ChunkKey, Element, IndexKeys, IndexKey>,
index_key: Cow<'a, IndexKey>,
) -> Self {
MatchingSecondaryIndex {
query,
secondary_index: secondary_index.clone(),
index_key,
}
}
}
impl<'a, Q, ChunkKey, ItemKey, Element, IndexKeys, IndexKey> Query<ChunkKey, ItemKey, Element>
for MatchingSecondaryIndex<'a, Q, ChunkKey, Element, IndexKeys, IndexKey>
where
ChunkKey: BorrowedKey + ?Sized,
ChunkKey::Owned: ValidKey,
ItemKey: BorrowedKey + ?Sized,
ItemKey::Owned: ValidKey,
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
Element: Record<ChunkKey, ItemKey>,
for<'k> IndexKeys: Clone + Debug + Default + Eq + KeySet<'k, IndexKey>,
Q: Query<ChunkKey, ItemKey, Element> + Clone,
{
type ChunkIdxSet = Q::ChunkIdxSet;
type ItemIdxSet = Intersection<Q::ItemIdxSet, Option<Bitset>>;
fn chunk_idxs(&self, storage: &Storage<ChunkKey, ItemKey, Element>) -> Self::ChunkIdxSet {
let mut secondary_index_impl = self.secondary_index.0.write().unwrap();
assert_eq!(secondary_index_impl.parent_id, storage.id(), "Id mismatch: a secondary index may only be used with it's parent Storage, never any other Storage");
let result = self.query.chunk_idxs(storage);
secondary_index_impl.gc(storage);
for idx in result.clone().into_idx_iter().flatten() {
let chunk_key = secondary_index_impl.gc_chunk_list[idx]
.as_ref()
.cloned()
.expect("gc_chunk_list should not contain None immediately after gc");
secondary_index_impl.update_chunk(chunk_key.borrow(), &storage.internal_rvec()[idx]);
}
result
}
fn item_idxs(
&self,
chunk_key: &ChunkKey,
chunk_storage: &ChunkStorage<ChunkKey, ItemKey, Element>,
) -> Self::ItemIdxSet {
let secondary_index_impl = self.secondary_index.0.read().unwrap();
let parent_idxs = self.query.item_idxs(chunk_key, chunk_storage);
let ours_idxs: Option<Bitset> = secondary_index_impl
.index
.get(chunk_key)
.and_then(|map_summarize| {
map_summarize
.peek()
.reverse_index
.get(self.index_key.borrow())
})
.cloned();
IdxSet::intersection(parent_idxs, ours_idxs)
}
fn test(&self, element: &Element) -> bool {
self.query.test(element)
}
}
#[cfg(test)]
mod test {
#[test]
fn test_kitten_removal() {
use crate::queries::everything::*;
use crate::queries::secondary_index::*;
use std::borrow::Cow;
struct Kitten {
name: String,
colors: HashSet<String>,
}
impl Record<(), str> for Kitten {
fn chunk_key(self: &Kitten) -> Cow<()> {
Cow::Owned(())
}
fn item_key(self: &Kitten) -> Cow<str> {
Cow::Borrowed(&self.name)
}
}
let mut storage: Storage<(), str, Kitten> = Storage::new();
let by_color: SecondaryIndex<(), Kitten, HashSet<String>, str> =
SecondaryIndex::new(&storage, |kitten: &Kitten| Cow::Borrowed(&kitten.colors));
storage.add(Kitten {
name: String::from("mittens"),
colors: vec![String::from("black"), String::from("white")]
.into_iter()
.collect(),
});
storage.add(Kitten {
name: String::from("furball"),
colors: vec![String::from("orange")].into_iter().collect(),
});
storage.add(Kitten {
name: String::from("midnight"),
colors: vec![String::from("black")].into_iter().collect(),
});
storage.validate();
by_color.validate(&storage);
storage.remove(
&Everything.matching(&by_color, Cow::Borrowed("orange")),
std::mem::drop,
);
storage.validate();
by_color.validate(&storage);
storage.remove(
&Everything.matching(&by_color, Cow::Borrowed("white")),
std::mem::drop,
);
storage.validate();
by_color.validate(&storage);
storage.remove(
&Everything.matching(&by_color, Cow::Borrowed("black")),
std::mem::drop,
);
storage.validate();
by_color.validate(&storage);
}
}
impl<IndexKey> Default for ChunkSecondaryIndex<IndexKey>
where
IndexKey: BorrowedKey + ?Sized,
IndexKey::Owned: ValidKey,
{
fn default() -> Self {
ChunkSecondaryIndex {
reverse_index: HashMap::default(),
}
}
}
pub trait KeySet<'a, K>
where
K: ToOwned + ?Sized + 'a,
{
type KeySetIter: Iterator<Item = Cow<'a, K>>;
fn iter_keys(&'a self) -> Self::KeySetIter;
}
impl<'a, K, T> KeySet<'a, K> for [T; 1]
where
T: Clone + Borrow<K> + 'a,
K: ToOwned<Owned = T> + ?Sized + 'a,
{
#[allow(clippy::type_complexity)]
type KeySetIter = Map<<&'a [T] as IntoIterator>::IntoIter, fn(&'a T) -> Cow<'a, K>>;
fn iter_keys(&'a self) -> Self::KeySetIter {
self.iter().map(|t| Cow::Borrowed(t.borrow()))
}
}
impl<'a, K, T> KeySet<'a, K> for Option<T>
where
T: Clone + Borrow<K> + 'a,
K: ToOwned<Owned = T> + ?Sized + 'a,
{
#[allow(clippy::type_complexity)]
type KeySetIter = Map<<&'a Option<T> as IntoIterator>::IntoIter, fn(&'a T) -> Cow<'a, K>>;
fn iter_keys(&'a self) -> Self::KeySetIter {
self.iter().map(|t: &'a T| Cow::Borrowed(t.borrow()))
}
}
impl<'a, K, T, S> KeySet<'a, K> for HashSet<T, S>
where
T: Clone + Borrow<K> + 'a,
K: ToOwned<Owned = T> + ?Sized + 'a,
{
#[allow(clippy::type_complexity)]
type KeySetIter = Map<<&'a HashSet<T, S> as IntoIterator>::IntoIter, fn(&'a T) -> Cow<'a, K>>;
fn iter_keys(&'a self) -> Self::KeySetIter {
self.iter().map(|t: &'a T| Cow::Borrowed(t.borrow()))
}
}
impl<'a, K, T> KeySet<'a, K> for BTreeSet<T>
where
T: Clone + Borrow<K> + 'a,
K: ToOwned<Owned = T> + ?Sized + 'a,
{
#[allow(clippy::type_complexity)]
type KeySetIter = Map<<&'a BTreeSet<T> as IntoIterator>::IntoIter, fn(&'a T) -> Cow<'a, K>>;
fn iter_keys(&'a self) -> Self::KeySetIter {
self.iter().map(|t: &'a T| Cow::Borrowed(t.borrow()))
}
}