diff --git a/constellation/src/storage/mem_store.rs b/constellation/src/storage/mem_store.rs index f122590..acadf5f 100644 --- a/constellation/src/storage/mem_store.rs +++ b/constellation/src/storage/mem_store.rs @@ -158,7 +158,7 @@ impl LinkReader for MemStorage { HashSet::from_iter(filter_to_targets.iter().map(|s| Target::new(s))); let mut grouped_counts: HashMap)> = HashMap::new(); - for (did, rkey) in linkers.into_iter().cloned().filter_map(|l| l) { + for (did, rkey) in linkers.iter().flatten().cloned() { if !filter_dids.is_empty() && !filter_dids.contains(&did) { continue; } @@ -171,7 +171,7 @@ impl LinkReader for MemStorage { rkey, }) .unwrap_or(&Vec::new()) - .into_iter() + .iter() .filter_map(|(path, target)| { if *path == path_to_other && (filter_to_targets.is_empty() || filter_to_targets.contains(target)) diff --git a/constellation/src/storage/mod.rs b/constellation/src/storage/mod.rs index d48beaf..851aea8 100644 --- a/constellation/src/storage/mod.rs +++ b/constellation/src/storage/mod.rs @@ -59,6 +59,7 @@ pub trait LinkStorage: Send + Sync { } pub trait LinkReader: Clone + Send + Sync + 'static { + #[allow(clippy::too_many_arguments)] fn get_many_to_many_counts( &self, target: &str, diff --git a/constellation/src/storage/rocks_store.rs b/constellation/src/storage/rocks_store.rs index 76d2b26..94240b4 100644 --- a/constellation/src/storage/rocks_store.rs +++ b/constellation/src/storage/rocks_store.rs @@ -59,8 +59,8 @@ fn get_db_read_opts() -> Options { #[derive(Debug, Clone)] pub struct RocksStorage { pub db: Arc>, // TODO: mov seqs here (concat merge op will be fun) - did_id_table: IdTable, - target_id_table: IdTable, + did_id_table: IdTable, + target_id_table: IdTable, is_writer: bool, backup_task: Arc>>>, } @@ -88,10 +88,7 @@ where fn cf_descriptor(&self) -> ColumnFamilyDescriptor { ColumnFamilyDescriptor::new(&self.name, rocks_opts_base()) } - fn init( - self, - db: &DBWithThreadMode, - ) -> Result> { + fn init(self, db: &DBWithThreadMode) -> Result> { if db.cf_handle(&self.name).is_none() { bail!("failed to get cf handle from db -- was the db open with our .cf_descriptor()?"); } @@ -122,7 +119,7 @@ where } } #[derive(Debug, Clone)] -struct IdTable +struct IdTable where Orig: KeyFromRocks, for<'a> &'a Orig: AsRocksKey, @@ -130,7 +127,7 @@ where base: IdTableBase, priv_id_seq: u64, } -impl IdTable +impl IdTable where Orig: KeyFromRocks, for<'v> &'v IdVal: AsRocksValue, @@ -181,16 +178,11 @@ where id_value })) } + fn estimate_count(&self) -> u64 { self.base.id_seq.load(Ordering::SeqCst) - 1 // -1 because seq zero is reserved } -} -impl IdTable -where - Orig: KeyFromRocks, - for<'v> &'v IdVal: AsRocksValue, - for<'k> &'k Orig: AsRocksKey, -{ + fn get_or_create_id_val( &mut self, db: &DBWithThreadMode, @@ -218,22 +210,6 @@ where } } } -impl IdTable -where - Orig: KeyFromRocks, - for<'v> &'v IdVal: AsRocksValue, - for<'k> &'k Orig: AsRocksKey, -{ - fn get_or_create_id_val( - &mut self, - db: &DBWithThreadMode, - batch: &mut WriteBatch, - orig: &Orig, - ) -> Result { - let cf = db.cf_handle(&self.base.name).unwrap(); - self.__get_or_create_id_val(&cf, db, batch, orig) - } -} impl IdTableValue for DidIdValue { fn new(v: u64) -> Self { @@ -263,8 +239,8 @@ impl RocksStorage { } fn open_readmode(path: impl AsRef, readonly: bool) -> Result { - let did_id_table = IdTable::<_, _, true>::setup(DID_IDS_CF); - let target_id_table = IdTable::<_, _, true>::setup(TARGET_IDS_CF); + let did_id_table = IdTable::setup(DID_IDS_CF); + let target_id_table = IdTable::setup(TARGET_IDS_CF); let cfs = vec![ // id reference tables @@ -857,24 +833,27 @@ impl LinkReader for RocksStorage { }; let filter_did_ids: HashMap = filter_dids - .into_iter() + .iter() .filter_map(|did| self.did_id_table.get_id_val(&self.db, did).transpose()) .collect::>>()? .into_iter() .map(|DidIdValue(id, active)| (id, active)) .collect(); - let filter_to_target_ids = filter_to_targets - .into_iter() - .filter_map(|target| { - self.target_id_table - .get_id_val( - &self.db, - &TargetKey(Target(target.to_string()), collection.clone(), path.clone()), - ) - .transpose() - }) - .collect::>>()?; + // stored targets are keyed by triples of (target, collection, path). + // target filtering only consideres the target itself, so we actually + // need to do a prefix iteration of all target ids for this target and + // keep them all. + // i *think* the number of keys at a target prefix should usually be + // pretty small, so this is hopefully fine. but if it turns out to be + // large, we can push this filtering back into the main links loop and + // do forward db queries per backlink to get the raw target back out. + let mut filter_to_target_ids: HashSet = HashSet::new(); + for t in filter_to_targets { + for (_, target_id) in self.iter_targets_for_target(&Target(t.to_string())) { + filter_to_target_ids.insert(target_id); + } + } let linkers = self.get_target_linkers(&target_id)?; @@ -923,7 +902,7 @@ impl LinkReader for RocksStorage { // (this check continues after the did-lookup, which we have to do) let page_is_full = grouped_counts.len() as u64 >= limit; if page_is_full { - let current_max = grouped_counts.keys().rev().next().unwrap(); // limit should be non-zero bleh + let current_max = grouped_counts.keys().next_back().unwrap(); // limit should be non-zero bleh if fwd_target > *current_max { continue; } @@ -952,7 +931,7 @@ impl LinkReader for RocksStorage { Default::default() }); entry.0 += 1; - entry.1.insert(did_id.clone()); + entry.1.insert(did_id); if should_evict { grouped_counts.pop_last(); @@ -961,19 +940,21 @@ impl LinkReader for RocksStorage { let mut items: Vec<(String, u64, u64)> = Vec::with_capacity(grouped_counts.len()); for (target_id, (n, dids)) in &grouped_counts { - let Some(target) = self.target_id_table.get_val_from_id(&self.db, target_id.0)? else { + let Some(target) = self + .target_id_table + .get_val_from_id(&self.db, target_id.0)? + else { eprintln!("failed to look up target from target_id {target_id:?}"); continue; }; - items.push((target.0.0, *n, dids.len() as u64)); + items.push((target.0 .0, *n, dids.len() as u64)); } let next = if grouped_counts.len() as u64 >= limit { // yeah.... it's a number saved as a string......sorry grouped_counts .keys() - .rev() - .next() + .next_back() .map(|k| format!("{}", k.0)) } else { None