diff --git a/Cargo.lock b/Cargo.lock
index b07178967..6865c19ce 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -354,6 +354,12 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "567b077b825e468cc974f0020d4082ee6e03132512f207ef1a02fd5d00d1f32d"
 
+[[package]]
+name = "ahash"
+version = "0.3.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e8fd72866655d1904d6b0997d0b07ba561047d070fbe29de039031c641b61217"
+
 [[package]]
 name = "aho-corasick"
 version = "0.7.10"
@@ -1294,6 +1300,16 @@ version = "1.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d36fab90f82edc3c747f9d438e06cf0a491055896f2a279638bb5beed6c40177"
 
+[[package]]
+name = "hashbrown"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ab9b7860757ce258c89fd48d28b68c41713e597a7b09e793f6c6a6e2ea37c827"
+dependencies = [
+ "ahash",
+ "autocfg",
+]
+
 [[package]]
 name = "heck"
 version = "0.3.1"
@@ -1518,6 +1534,7 @@ dependencies = [
  "env_logger",
  "futures",
  "futures-util",
+ "hashbrown",
  "idlset",
  "kanidm_proto",
  "lazy_static",
diff --git a/kanidm_client/tests/proto_v1_test.rs b/kanidm_client/tests/proto_v1_test.rs
index 2461ffaff..04c401526 100644
--- a/kanidm_client/tests/proto_v1_test.rs
+++ b/kanidm_client/tests/proto_v1_test.rs
@@ -763,6 +763,9 @@ fn test_server_rest_totp_auth_lifecycle() {
                     .unwrap(),
             )
             .expect("Failed to do totp?");
+        // TODO: It's extremely rare, but it's happened ONCE where, the time window
+        // elapsed DURING this test, so there is a minor possibility of this actually
+        // having a false negative. Is it possible to prevent this?
         assert!(rsclient_good
             .auth_password_totp("demo_account", "sohdi3iuHo6mai7noh0a", totp)
             .is_ok());
diff --git a/kanidm_proto/src/v1.rs b/kanidm_proto/src/v1.rs
index 092289c96..68f20f68d 100644
--- a/kanidm_proto/src/v1.rs
+++ b/kanidm_proto/src/v1.rs
@@ -293,7 +293,7 @@ impl fmt::Display for Entry {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize, Clone, Ord, PartialOrd, Eq, PartialEq)]
+#[derive(Debug, Serialize, Deserialize, Clone, Ord, PartialOrd, Eq, PartialEq, Hash)]
 #[serde(rename_all = "lowercase")]
 pub enum Filter {
     // This is attr - value
diff --git a/kanidmd/Cargo.toml b/kanidmd/Cargo.toml
index 1e0613120..9c0d25371 100644
--- a/kanidmd/Cargo.toml
+++ b/kanidmd/Cargo.toml
@@ -60,6 +60,7 @@ r2d2_sqlite = "0.14"
 structopt = { version = "0.3", default-features = false }
 time = "0.1"
 
+hashbrown = "0.8"
 concread = "0.1"
 # concread = { path = "../../concread" }
 crossbeam = "0.7"
diff --git a/kanidmd/src/lib/access.rs b/kanidmd/src/lib/access.rs
index e7ef7f97e..3b82bee6a 100644
--- a/kanidmd/src/lib/access.rs
+++ b/kanidmd/src/lib/access.rs
@@ -19,9 +19,8 @@
 use concread::cowcell::*;
 use kanidm_proto::v1::Filter as ProtoFilter;
 use kanidm_proto::v1::OperationError;
-// use std::collections::BTreeMap;
 use std::collections::BTreeSet;
-// use std::collections::HashSet;
+// use hashbrown::HashSet;
 use std::ops::DerefMut;
 use uuid::Uuid;
 
@@ -355,7 +354,6 @@ impl AccessControlProfile {
 
 #[derive(Clone)]
 struct AccessControlsInner {
-    // acps_search: BTreeMap<Uuid, AccessControlSearch>,
     acps_search: Vec<AccessControlSearch>,
     acps_create: Vec<AccessControlCreate>,
     acps_modify: Vec<AccessControlModify>,
@@ -367,7 +365,6 @@ pub struct AccessControls {
 }
 
 pub trait AccessControlsTransaction {
-    // fn get_search(&self) -> &BTreeMap<Uuid, AccessControlSearch>;
     fn get_search(&self) -> &Vec<AccessControlSearch>;
     fn get_create(&self) -> &Vec<AccessControlCreate>;
     fn get_modify(&self) -> &Vec<AccessControlModify>;
@@ -1244,7 +1241,6 @@ impl<'a> AccessControlsWriteTransaction<'a> {
 }
 
 impl<'a> AccessControlsTransaction for AccessControlsWriteTransaction<'a> {
-    // fn get_search(&self) -> &BTreeMap<Uuid, AccessControlSearch> {
     fn get_search(&self) -> &Vec<AccessControlSearch> {
         &self.inner.acps_search
     }
@@ -1271,7 +1267,6 @@ pub struct AccessControlsReadTransaction {
 }
 
 impl AccessControlsTransaction for AccessControlsReadTransaction {
-    // fn get_search(&self) -> &BTreeMap<Uuid, AccessControlSearch> {
     fn get_search(&self) -> &Vec<AccessControlSearch> {
         &self.inner.acps_search
     }
diff --git a/kanidmd/src/lib/be/mod.rs b/kanidmd/src/lib/be/mod.rs
index 593c28f69..b6efeceda 100644
--- a/kanidmd/src/lib/be/mod.rs
+++ b/kanidmd/src/lib/be/mod.rs
@@ -2,8 +2,8 @@ use std::convert::TryFrom;
 use std::fs;
 
 use crate::value::IndexType;
+use hashbrown::HashSet as Set;
 use std::cell::UnsafeCell;
-use std::collections::HashSet as Set;
 use std::sync::Arc;
 
 use crate::audit::AuditScope;
@@ -102,6 +102,7 @@ pub trait BackendTransaction {
 
     /// Recursively apply a filter, transforming into IDL's on the way. This builds a query
     /// execution log, so that it can be examined how an operation proceeded.
+    #[allow(clippy::cognitive_complexity)]
     fn filter2idl(
         &self,
         au: &mut AuditScope,
@@ -413,6 +414,44 @@ pub trait BackendTransaction {
                 // debug!("final cand set ==> {:?}", cand_idl);
                 (cand_idl, setplan)
             } // end and
+            FilterResolved::Inclusion(l) => {
+                // For inclusion to be valid, every term must have *at least* one element present.
+                // This really relies on indexing, and so it's internal only - generally only
+                // for fully indexed existance queries, such as from refint.
+
+                // This has a lot in common with an And and Or but not really quite either.
+                let mut plan = Vec::new();
+                let mut result = IDLBitRange::new();
+                // For each filter in l
+                for f in l.iter() {
+                    // get their idls
+                    match self.filter2idl(au, f, thres)? {
+                        (IDL::Indexed(idl), fp) => {
+                            plan.push(fp);
+                            if idl.is_empty() {
+                                // It's empty, so something is missing. Bail fast.
+                                lfilter!(au, "Inclusion is unable to proceed - an empty (missing) item was found!");
+                                let setplan = FilterPlan::InclusionIndexed(plan);
+                                return Ok((IDL::Indexed(IDLBitRange::new()), setplan));
+                            } else {
+                                result = result | idl;
+                            }
+                        }
+                        (_, fp) => {
+                            plan.push(fp);
+                            lfilter_error!(
+                                au,
+                                "Inclusion is unable to proceed - all terms must be fully indexed!"
+                            );
+                            let setplan = FilterPlan::InclusionInvalid(plan);
+                            return Ok((IDL::Partial(IDLBitRange::new()), setplan));
+                        }
+                    }
+                } // end or.iter()
+                  // If we got here, every term must have been indexed
+                let setplan = FilterPlan::InclusionIndexed(plan);
+                (IDL::Indexed(result), setplan)
+            }
             // So why does this return empty? Normally we actually process an AndNot in the context
             // of an "AND" query, but if it's used anywhere else IE the root filter, then there is
             // no other set to exclude - therefore it's empty set. Additionally, even in an OR query
@@ -1312,8 +1351,8 @@ impl Backend {
 
 #[cfg(test)]
 mod tests {
+    use hashbrown::HashSet as Set;
     use idlset::IDLBitRange;
-    use std::collections::HashSet as Set;
     use std::fs;
     use std::iter::FromIterator;
     use uuid::Uuid;
@@ -1337,7 +1376,7 @@ mod tests {
             let mut audit = AuditScope::new("run_test", uuid::Uuid::new_v4(), None);
 
             // This is a demo idxmeta, purely for testing.
-            let mut idxmeta = Set::new();
+            let mut idxmeta = Set::with_capacity(16);
             idxmeta.insert(IdxKey {
                 attr: "name".to_string(),
                 itype: IndexType::EQUALITY,
@@ -1428,8 +1467,8 @@ mod tests {
             assert_eq!(empty_result, Err(OperationError::EmptyRequest));
 
             let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-            e.add_ava("userid", &Value::from("william"));
-            e.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e.add_ava("userid", Value::from("william"));
+            e.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e = unsafe { e.into_sealed_new() };
 
             let single_result = be.create(audit, vec![e.clone()]);
@@ -1447,8 +1486,8 @@ mod tests {
             ltrace!(audit, "Simple Search");
 
             let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-            e.add_ava("userid", &Value::from("claire"));
-            e.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e.add_ava("userid", Value::from("claire"));
+            e.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e = unsafe { e.into_sealed_new() };
 
             let single_result = be.create(audit, vec![e.clone()]);
@@ -1475,12 +1514,12 @@ mod tests {
             ltrace!(audit, "Simple Modify");
             // First create some entries (3?)
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("userid", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("userid", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("userid", &Value::from("alice"));
-            e2.add_ava("uuid", &Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
+            e2.add_ava("userid", Value::from("alice"));
+            e2.add_ava("uuid", Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
 
             let ve1 = unsafe { e1.clone().into_sealed_new() };
             let ve2 = unsafe { e2.clone().into_sealed_new() };
@@ -1512,8 +1551,8 @@ mod tests {
             // Make some changes to r1, r2.
             let pre1 = unsafe { r1.clone().into_sealed_committed() };
             let pre2 = unsafe { r2.clone().into_sealed_committed() };
-            r1.add_ava("desc", &Value::from("modified"));
-            r2.add_ava("desc", &Value::from("modified"));
+            r1.add_ava("desc", Value::from("modified"));
+            r2.add_ava("desc", Value::from("modified"));
 
             // Now ... cheat.
 
@@ -1549,16 +1588,16 @@ mod tests {
 
             // First create some entries (3?)
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("userid", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("userid", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("userid", &Value::from("alice"));
-            e2.add_ava("uuid", &Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
+            e2.add_ava("userid", Value::from("alice"));
+            e2.add_ava("uuid", Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
 
             let mut e3: Entry<EntryInit, EntryNew> = Entry::new();
-            e3.add_ava("userid", &Value::from("lucy"));
-            e3.add_ava("uuid", &Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
+            e3.add_ava("userid", Value::from("lucy"));
+            e3.add_ava("uuid", Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
 
             let ve1 = unsafe { e1.clone().into_sealed_new() };
             let ve2 = unsafe { e2.clone().into_sealed_new() };
@@ -1590,8 +1629,8 @@ mod tests {
             // WARNING: Normally, this isn't possible, but we are pursposefully breaking
             // the state machine rules here!!!!
             let mut e4: Entry<EntryInit, EntryNew> = Entry::new();
-            e4.add_ava("userid", &Value::from("amy"));
-            e4.add_ava("uuid", &Value::from("21d816b5-1f6a-4696-b7c1-6ed06d22ed81"));
+            e4.add_ava("userid", Value::from("amy"));
+            e4.add_ava("uuid", Value::from("21d816b5-1f6a-4696-b7c1-6ed06d22ed81"));
 
             let ve4 = unsafe { e4.clone().into_sealed_committed() };
 
@@ -1619,16 +1658,16 @@ mod tests {
         run_test!(|audit: &mut AuditScope, be: &mut BackendWriteTransaction| {
             // First create some entries (3?)
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("userid", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("userid", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("userid", &Value::from("alice"));
-            e2.add_ava("uuid", &Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
+            e2.add_ava("userid", Value::from("alice"));
+            e2.add_ava("uuid", Value::from("4b6228ab-1dbe-42a4-a9f5-f6368222438e"));
 
             let mut e3: Entry<EntryInit, EntryNew> = Entry::new();
-            e3.add_ava("userid", &Value::from("lucy"));
-            e3.add_ava("uuid", &Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
+            e3.add_ava("userid", Value::from("lucy"));
+            e3.add_ava("uuid", Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
 
             let ve1 = unsafe { e1.clone().into_sealed_new() };
             let ve2 = unsafe { e2.clone().into_sealed_new() };
@@ -1693,13 +1732,13 @@ mod tests {
         run_test!(|audit: &mut AuditScope, be: &mut BackendWriteTransaction| {
             // Add some test data?
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::new_iname_s("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("name", Value::new_iname_s("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("name", &Value::new_iname_s("claire"));
-            e2.add_ava("uuid", &Value::from("bd651620-00dd-426b-aaa0-4494f7b7906f"));
+            e2.add_ava("name", Value::new_iname_s("claire"));
+            e2.add_ava("uuid", Value::from("bd651620-00dd-426b-aaa0-4494f7b7906f"));
             let e2 = unsafe { e2.into_sealed_new() };
 
             be.create(audit, vec![e1.clone(), e2.clone()]).unwrap();
@@ -1823,8 +1862,8 @@ mod tests {
             // Test that on entry create, the indexes are made correctly.
             // this is a similar case to reindex.
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("name", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let rset = be.create(audit, vec![e1.clone()]).unwrap();
@@ -1910,18 +1949,18 @@ mod tests {
             // Test that on entry create, the indexes are made correctly.
             // this is a similar case to reindex.
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("name", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("name", &Value::from("claire"));
-            e2.add_ava("uuid", &Value::from("bd651620-00dd-426b-aaa0-4494f7b7906f"));
+            e2.add_ava("name", Value::from("claire"));
+            e2.add_ava("uuid", Value::from("bd651620-00dd-426b-aaa0-4494f7b7906f"));
             let e2 = unsafe { e2.into_sealed_new() };
 
             let mut e3: Entry<EntryInit, EntryNew> = Entry::new();
-            e3.add_ava("userid", &Value::from("lucy"));
-            e3.add_ava("uuid", &Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
+            e3.add_ava("userid", Value::from("lucy"));
+            e3.add_ava("uuid", Value::from("7b23c99d-c06b-4a9a-a958-3afa56383e1d"));
             let e3 = unsafe { e3.into_sealed_new() };
 
             let mut rset = be
@@ -1980,21 +2019,21 @@ mod tests {
             // us. For the test to be "accurate" we must add one attr, remove one attr
             // and change one attr.
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
-            e1.add_ava("ta", &Value::from("test"));
+            e1.add_ava("name", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("ta", Value::from("test"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let rset = be.create(audit, vec![e1.clone()]).unwrap();
             // Now, alter the new entry.
             let mut ce1 = unsafe { rset[0].clone().into_invalid() };
             // add something.
-            ce1.add_ava("tb", &Value::from("test"));
+            ce1.add_ava("tb", Value::from("test"));
             // remove something.
             ce1.purge_ava("ta");
             // mod something.
             ce1.purge_ava("name");
-            ce1.add_ava("name", &Value::from("claire"));
+            ce1.add_ava("name", Value::from("claire"));
 
             let ce1 = unsafe { ce1.into_sealed_committed() };
 
@@ -2033,8 +2072,8 @@ mod tests {
             // This will be needing to be correct for conflicts when we add
             // replication support!
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("name", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let rset = be.create(audit, vec![e1.clone()]).unwrap();
@@ -2042,8 +2081,8 @@ mod tests {
             let mut ce1 = unsafe { rset[0].clone().into_invalid() };
             ce1.purge_ava("name");
             ce1.purge_ava("uuid");
-            ce1.add_ava("name", &Value::from("claire"));
-            ce1.add_ava("uuid", &Value::from("04091a7a-6ce4-42d2-abf5-c2ce244ac9e8"));
+            ce1.add_ava("name", Value::from("claire"));
+            ce1.add_ava("uuid", Value::from("04091a7a-6ce4-42d2-abf5-c2ce244ac9e8"));
             let ce1 = unsafe { ce1.into_sealed_committed() };
 
             be.modify(audit, &rset, &vec![ce1]).unwrap();
@@ -2104,15 +2143,15 @@ mod tests {
 
             // Create a test entry with some indexed / unindexed values.
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("name", &Value::from("william"));
-            e1.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
-            e1.add_ava("no-index", &Value::from("william"));
-            e1.add_ava("other-no-index", &Value::from("william"));
+            e1.add_ava("name", Value::from("william"));
+            e1.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d1"));
+            e1.add_ava("no-index", Value::from("william"));
+            e1.add_ava("other-no-index", Value::from("william"));
             let e1 = unsafe { e1.into_sealed_new() };
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("name", &Value::from("claire"));
-            e2.add_ava("uuid", &Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d2"));
+            e2.add_ava("name", Value::from("claire"));
+            e2.add_ava("uuid", Value::from("db237e8a-0079-4b8c-8a56-593b22aa44d2"));
             let e2 = unsafe { e2.into_sealed_new() };
 
             let _rset = be.create(audit, vec![e1.clone(), e2.clone()]).unwrap();
diff --git a/kanidmd/src/lib/constants/mod.rs b/kanidmd/src/lib/constants/mod.rs
index 3a36d4f2d..8fc725cf4 100644
--- a/kanidmd/src/lib/constants/mod.rs
+++ b/kanidmd/src/lib/constants/mod.rs
@@ -13,7 +13,7 @@ pub use crate::constants::system_config::*;
 pub use crate::constants::uuids::*;
 
 // Increment this as we add new schema types and values!!!
-pub const SYSTEM_INDEX_VERSION: i64 = 10;
+pub const SYSTEM_INDEX_VERSION: i64 = 11;
 // On test builds, define to 60 seconds
 #[cfg(test)]
 pub const PURGE_FREQUENCY: u64 = 60;
diff --git a/kanidmd/src/lib/constants/schema.rs b/kanidmd/src/lib/constants/schema.rs
index 3080d4880..0be817532 100644
--- a/kanidmd/src/lib/constants/schema.rs
+++ b/kanidmd/src/lib/constants/schema.rs
@@ -320,7 +320,7 @@ pub const JSON_SCHEMA_ATTR_BADLIST_PASSWORD: &str = r#"{
         "EQUALITY"
       ],
       "unique": [
-        "true"
+        "false"
       ],
       "multivalue": [
         "true"
diff --git a/kanidmd/src/lib/core/mod.rs b/kanidmd/src/lib/core/mod.rs
index 1180cc4ce..c7710cb16 100644
--- a/kanidmd/src/lib/core/mod.rs
+++ b/kanidmd/src/lib/core/mod.rs
@@ -427,7 +427,7 @@ async fn json_rest_event_credential_put(
 
 // Okay, so a put normally needs
 //  * filter of what we are working on (id + class)
-//  * a BTreeMap<String, Vec<String>> that we turn into a modlist.
+//  * a Map<String, Vec<String>> that we turn into a modlist.
 //
 // OR
 //  * filter of what we are working on (id + class)
diff --git a/kanidmd/src/lib/entry.rs b/kanidmd/src/lib/entry.rs
index fc8a5e2d5..27f12ff54 100644
--- a/kanidmd/src/lib/entry.rs
+++ b/kanidmd/src/lib/entry.rs
@@ -48,10 +48,9 @@ use crate::be::IdxKey;
 use ldap3_server::simple::{LdapPartialAttribute, LdapSearchResultEntry};
 use std::collections::BTreeSet as Set;
 use std::collections::BTreeSet;
-// BTreeMap could be faster, but it's small datasets?
-// use std::collections::HashMap as Map;
-use std::collections::BTreeMap as Map;
-use std::collections::HashSet;
+// use std::collections::BTreeMap as Map;
+use hashbrown::HashMap as Map;
+use hashbrown::HashSet;
 use uuid::Uuid;
 
 // use std::convert::TryFrom;
@@ -246,7 +245,7 @@ impl Entry<EntryInit, EntryNew> {
             // This means NEVER COMMITED
             valid: EntryInit,
             state: EntryNew,
-            attrs: Map::new(),
+            attrs: Map::with_capacity(32),
         }
     }
 
@@ -486,7 +485,7 @@ impl Entry<EntryInit, EntryNew> {
     }
 
     #[cfg(test)]
-    pub fn add_ava(&mut self, attr: &str, value: &Value) {
+    pub fn add_ava(&mut self, attr: &str, value: Value) {
         self.add_ava_int(attr, value)
     }
 }
@@ -721,7 +720,7 @@ impl Entry<EntryInvalid, EntryCommitted> {
     }
 
     pub fn into_recycled(mut self) -> Self {
-        self.add_ava("class", &Value::new_class("recycled"));
+        self.add_ava("class", Value::new_class("recycled"));
 
         Entry {
             valid: self.valid.clone(),
@@ -1475,19 +1474,12 @@ impl Entry<EntryReduced, EntryCommitted> {
 
 // impl<STATE> Entry<EntryValid, STATE> {
 impl<VALID, STATE> Entry<VALID, STATE> {
-    fn add_ava_int(&mut self, attr: &str, value: &Value) {
+    fn add_ava_int(&mut self, attr: &str, value: Value) {
         // How do we make this turn into an ok / err?
-        self.attrs
-            .entry(attr.to_string())
-            .and_modify(|v| {
-                // Here we need to actually do a check/binary search ...
-                if v.contains(value) {
-                    // It already exists, done!
-                } else {
-                    v.insert(value.clone());
-                }
-            })
-            .or_insert(btreeset![value.clone()]);
+        let v = self.attrs.entry(attr.to_string()).or_insert_with(Set::new);
+        // Here we need to actually do a check/binary search ...
+        v.insert(value);
+        // Doesn't matter if it already exists, equality will replace.
     }
 
     fn set_last_changed(&mut self, cid: Cid) {
@@ -1699,6 +1691,12 @@ impl<VALID, STATE> Entry<VALID, STATE> {
                     acc
                 }
             }),
+            FilterResolved::Inclusion(_) => {
+                // An inclusion doesn't make sense on an entry in isolation!
+                // Inclusions are part of exists queries, on search they mean
+                // nothing!
+                false
+            }
             FilterResolved::AndNot(f) => !self.entry_match_no_index_inner(f),
         }
     }
@@ -1796,7 +1794,10 @@ where
     // a list of syntax violations ...
     // If this already exists, we silently drop the event? Is that an
     // acceptable interface?
-    pub fn add_ava(&mut self, attr: &str, value: &Value) {
+    //
+    // TODO: This should take Value not &Value, would save a lot of clones
+    // around the codebase.
+    pub fn add_ava(&mut self, attr: &str, value: Value) {
         self.add_ava_int(attr, value)
     }
 
@@ -1809,6 +1810,15 @@ where
         });
     }
 
+    // Need something that can remove by difference?
+    pub(crate) fn remove_avas(&mut self, attr: &str, values: &BTreeSet<PartialValue>) {
+        if let Some(set) = self.attrs.get_mut(attr) {
+            values.iter().for_each(|k| {
+                set.remove(k);
+            })
+        }
+    }
+
     pub fn purge_ava(&mut self, attr: &str) {
         self.attrs.remove(attr);
     }
@@ -1817,13 +1827,6 @@ where
         self.attrs.remove(attr)
     }
 
-    /// Overwrite the existing avas.
-    pub fn set_avas(&mut self, attr: &str, values: Vec<Value>) {
-        // Overwrite the existing value, build a tree from the list.
-        let x: Set<_> = values.into_iter().collect();
-        let _ = self.attrs.insert(attr.to_string(), x);
-    }
-
     /// Provide a true ava set.
     pub fn set_ava(&mut self, attr: &str, values: Set<Value>) {
         // Overwrite the existing value, build a tree from the list.
@@ -1848,7 +1851,7 @@ where
         // mutate
         for modify in modlist {
             match modify {
-                Modify::Present(a, v) => self.add_ava(a.as_str(), v),
+                Modify::Present(a, v) => self.add_ava(a.as_str(), v.clone()),
                 Modify::Removed(a, v) => self.remove_ava(a.as_str(), v),
                 Modify::Purged(a) => self.purge_ava(a.as_str()),
             }
@@ -1888,7 +1891,7 @@ impl From<&SchemaAttribute> for Entry<EntryInit, EntryNew> {
         let syntax_v = btreeset![Value::from(s.syntax.clone())];
 
         // Build the Map of the attributes relevant
-        let mut attrs: Map<String, Set<Value>> = Map::new();
+        let mut attrs: Map<String, Set<Value>> = Map::with_capacity(16);
         attrs.insert("attributename".to_string(), name_v);
         attrs.insert("description".to_string(), desc_v);
         attrs.insert("uuid".to_string(), uuid_v);
@@ -1922,7 +1925,7 @@ impl From<&SchemaClass> for Entry<EntryInit, EntryNew> {
         let name_v = btreeset![Value::new_iutf8(s.name.clone())];
         let desc_v = btreeset![Value::new_utf8(s.description.clone())];
 
-        let mut attrs: Map<String, Set<Value>> = Map::new();
+        let mut attrs: Map<String, Set<Value>> = Map::with_capacity(16);
         attrs.insert("classname".to_string(), name_v);
         attrs.insert("description".to_string(), desc_v);
         attrs.insert("uuid".to_string(), uuid_v);
@@ -1969,14 +1972,14 @@ mod tests {
     use crate::entry::{Entry, EntryInit, EntryInvalid, EntryNew};
     use crate::modify::{Modify, ModifyList};
     use crate::value::{IndexType, PartialValue, Value};
+    use hashbrown::HashSet;
     use std::collections::BTreeSet as Set;
-    use std::collections::HashSet;
 
     #[test]
     fn test_entry_basic() {
         let mut e: Entry<EntryInit, EntryNew> = Entry::new();
 
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
     }
 
     #[test]
@@ -1988,8 +1991,8 @@ mod tests {
         // are adding ... Or do we validate after the changes are made in
         // total?
         let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-        e.add_ava("userid", &Value::from("william"));
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
 
         let values = e.get_ava_set("userid").expect("Failed to get ava");
         // Should only be one value!
@@ -1999,7 +2002,7 @@ mod tests {
     #[test]
     fn test_entry_pres() {
         let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
 
         assert!(e.attribute_pres("userid"));
         assert!(!e.attribute_pres("name"));
@@ -2009,7 +2012,7 @@ mod tests {
     fn test_entry_equality() {
         let mut e: Entry<EntryInit, EntryNew> = Entry::new();
 
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
 
         assert!(e.attribute_equality("userid", &PartialValue::new_utf8s("william")));
         assert!(!e.attribute_equality("userid", &PartialValue::new_utf8s("test")));
@@ -2022,7 +2025,7 @@ mod tests {
     fn test_entry_substring() {
         let mut e: Entry<EntryInit, EntryNew> = Entry::new();
 
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
 
         assert!(e.attribute_substring("userid", &PartialValue::new_utf8s("william")));
         assert!(e.attribute_substring("userid", &PartialValue::new_utf8s("will")));
@@ -2042,14 +2045,14 @@ mod tests {
         let pv10 = PartialValue::new_uint32(10);
         let pv15 = PartialValue::new_uint32(15);
 
-        e1.add_ava("a", &Value::new_uint32(10));
+        e1.add_ava("a", Value::new_uint32(10));
 
         assert!(e1.attribute_lessthan("a", &pv2) == false);
         assert!(e1.attribute_lessthan("a", &pv8) == false);
         assert!(e1.attribute_lessthan("a", &pv10) == false);
         assert!(e1.attribute_lessthan("a", &pv15) == true);
 
-        e1.add_ava("a", &Value::new_uint32(8));
+        e1.add_ava("a", Value::new_uint32(8));
 
         assert!(e1.attribute_lessthan("a", &pv2) == false);
         assert!(e1.attribute_lessthan("a", &pv8) == false);
@@ -2062,7 +2065,7 @@ mod tests {
         // Test application of changes to an entry.
         let mut e: Entry<EntryInvalid, EntryNew> = unsafe { Entry::new().into_invalid_new() };
 
-        e.add_ava("userid", &Value::from("william"));
+        e.add_ava("userid", Value::from("william"));
 
         let present_single_mods = unsafe {
             ModifyList::new_valid_list(vec![Modify::Present(
@@ -2132,18 +2135,18 @@ mod tests {
     #[test]
     fn test_entry_idx_diff() {
         let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-        e1.add_ava("userid", &Value::from("william"));
+        e1.add_ava("userid", Value::from("william"));
         let mut e1_mod = e1.clone();
-        e1_mod.add_ava("extra", &Value::from("test"));
+        e1_mod.add_ava("extra", Value::from("test"));
 
         let e1 = unsafe { e1.into_sealed_committed() };
         let e1_mod = unsafe { e1_mod.into_sealed_committed() };
 
         let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-        e2.add_ava("userid", &Value::from("claire"));
+        e2.add_ava("userid", Value::from("claire"));
         let e2 = unsafe { e2.into_sealed_committed() };
 
-        let mut idxmeta = HashSet::new();
+        let mut idxmeta = HashSet::with_capacity(8);
         idxmeta.insert(IdxKey {
             attr: "userid".to_string(),
             itype: IndexType::EQUALITY,
@@ -2244,18 +2247,18 @@ mod tests {
     #[test]
     fn test_entry_mask_recycled_ts() {
         let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-        e1.add_ava("class", &Value::new_class("person"));
+        e1.add_ava("class", Value::new_class("person"));
         let e1 = unsafe { e1.into_sealed_committed() };
         assert!(e1.mask_recycled_ts().is_some());
 
         let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-        e2.add_ava("class", &Value::new_class("person"));
-        e2.add_ava("class", &Value::new_class("recycled"));
+        e2.add_ava("class", Value::new_class("person"));
+        e2.add_ava("class", Value::new_class("recycled"));
         let e2 = unsafe { e2.into_sealed_committed() };
         assert!(e2.mask_recycled_ts().is_none());
 
         let mut e3: Entry<EntryInit, EntryNew> = Entry::new();
-        e3.add_ava("class", &Value::new_class("tombstone"));
+        e3.add_ava("class", Value::new_class("tombstone"));
         let e3 = unsafe { e3.into_sealed_committed() };
         assert!(e3.mask_recycled_ts().is_none());
     }
@@ -2269,7 +2272,7 @@ mod tests {
         // none, some - test adding an entry gives back add sets
         {
             let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-            e.add_ava("class", &Value::new_class("person"));
+            e.add_ava("class", Value::new_class("person"));
             let e = unsafe { e.into_sealed_committed() };
 
             assert!(Entry::idx_name2uuid_diff(None, Some(&e)) == (Some(Set::new()), None));
@@ -2277,13 +2280,13 @@ mod tests {
 
         {
             let mut e: Entry<EntryInit, EntryNew> = Entry::new();
-            e.add_ava("class", &Value::new_class("person"));
-            e.add_ava("gidnumber", &Value::new_uint32(1300));
-            e.add_ava("name", &Value::new_iname_s("testperson"));
-            e.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+            e.add_ava("class", Value::new_class("person"));
+            e.add_ava("gidnumber", Value::new_uint32(1300));
+            e.add_ava("name", Value::new_iname_s("testperson"));
+            e.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
             e.add_ava(
                 "uuid",
-                &Value::new_uuids("9fec0398-c46c-4df4-9df5-b0016f7d563f").unwrap(),
+                Value::new_uuids("9fec0398-c46c-4df4-9df5-b0016f7d563f").unwrap(),
             );
             let e = unsafe { e.into_sealed_committed() };
 
@@ -2323,14 +2326,14 @@ mod tests {
 
         {
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("class", &Value::new_class("person"));
-            e1.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+            e1.add_ava("class", Value::new_class("person"));
+            e1.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
             let e1 = unsafe { e1.into_sealed_committed() };
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("class", &Value::new_class("person"));
-            e2.add_ava("name", &Value::new_iname_s("testperson"));
-            e2.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+            e2.add_ava("class", Value::new_class("person"));
+            e2.add_ava("name", Value::new_iname_s("testperson"));
+            e2.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
             let e2 = unsafe { e2.into_sealed_committed() };
 
             // One attr added
@@ -2349,13 +2352,13 @@ mod tests {
         // Value changed, remove old, add new.
         {
             let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-            e1.add_ava("class", &Value::new_class("person"));
-            e1.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+            e1.add_ava("class", Value::new_class("person"));
+            e1.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
             let e1 = unsafe { e1.into_sealed_committed() };
 
             let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-            e2.add_ava("class", &Value::new_class("person"));
-            e2.add_ava("spn", &Value::new_spn_str("renameperson", "example.com"));
+            e2.add_ava("class", Value::new_class("person"));
+            e2.add_ava("spn", Value::new_spn_str("renameperson", "example.com"));
             let e2 = unsafe { e2.into_sealed_committed() };
 
             assert!(
@@ -2373,11 +2376,11 @@ mod tests {
         assert!(Entry::idx_uuid2spn_diff(None, None) == None);
 
         let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-        e1.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+        e1.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
         let e1 = unsafe { e1.into_sealed_committed() };
 
         let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-        e2.add_ava("spn", &Value::new_spn_str("renameperson", "example.com"));
+        e2.add_ava("spn", Value::new_spn_str("renameperson", "example.com"));
         let e2 = unsafe { e2.into_sealed_committed() };
 
         assert!(
@@ -2397,11 +2400,11 @@ mod tests {
         assert!(Entry::idx_uuid2rdn_diff(None, None) == None);
 
         let mut e1: Entry<EntryInit, EntryNew> = Entry::new();
-        e1.add_ava("spn", &Value::new_spn_str("testperson", "example.com"));
+        e1.add_ava("spn", Value::new_spn_str("testperson", "example.com"));
         let e1 = unsafe { e1.into_sealed_committed() };
 
         let mut e2: Entry<EntryInit, EntryNew> = Entry::new();
-        e2.add_ava("spn", &Value::new_spn_str("renameperson", "example.com"));
+        e2.add_ava("spn", Value::new_spn_str("renameperson", "example.com"));
         let e2 = unsafe { e2.into_sealed_committed() };
 
         assert!(
diff --git a/kanidmd/src/lib/filter.rs b/kanidmd/src/lib/filter.rs
index 233f5df78..1c0b6819b 100644
--- a/kanidmd/src/lib/filter.rs
+++ b/kanidmd/src/lib/filter.rs
@@ -17,12 +17,12 @@ use crate::server::{
     QueryServerReadTransaction, QueryServerTransaction, QueryServerWriteTransaction,
 };
 use crate::value::{IndexType, PartialValue};
+use hashbrown::HashSet;
 use kanidm_proto::v1::Filter as ProtoFilter;
 use kanidm_proto::v1::{OperationError, SchemaError};
 use ldap3_server::simple::LdapFilter;
 use std::cmp::{Ordering, PartialOrd};
 use std::collections::BTreeSet;
-use std::collections::HashSet;
 use std::iter;
 
 use uuid::Uuid;
@@ -61,6 +61,11 @@ pub fn f_and(vs: Vec<FC>) -> FC {
     FC::And(vs)
 }
 
+#[allow(dead_code)]
+pub fn f_inc(vs: Vec<FC>) -> FC {
+    FC::Inclusion(vs)
+}
+
 #[allow(dead_code)]
 pub fn f_andnot(fc: FC) -> FC {
     FC::AndNot(Box::new(fc))
@@ -107,6 +112,7 @@ pub enum FC<'a> {
     LessThan(&'a str, PartialValue),
     Or(Vec<FC<'a>>),
     And(Vec<FC<'a>>),
+    Inclusion(Vec<FC<'a>>),
     AndNot(Box<FC<'a>>),
     SelfUUID,
     // Not(Box<FC>),
@@ -122,6 +128,7 @@ enum FilterComp {
     LessThan(String, PartialValue),
     Or(Vec<FilterComp>),
     And(Vec<FilterComp>),
+    Inclusion(Vec<FilterComp>),
     AndNot(Box<FilterComp>),
     SelfUUID,
     // Does this mean we can add a true not to the type now?
@@ -141,6 +148,8 @@ pub enum FilterResolved {
     LessThan(String, PartialValue, bool),
     Or(Vec<FilterResolved>),
     And(Vec<FilterResolved>),
+    // All terms must have 1 or more items, or the inclusion is false!
+    Inclusion(Vec<FilterResolved>),
     AndNot(Box<FilterResolved>),
 }
 
@@ -182,6 +191,8 @@ pub enum FilterPlan {
     AndPartial(Vec<FilterPlan>),
     AndPartialThreshold(Vec<FilterPlan>),
     AndNot(Box<FilterPlan>),
+    InclusionInvalid(Vec<FilterPlan>),
+    InclusionIndexed(Vec<FilterPlan>),
 }
 
 /// A `Filter` is a logical set of assertions about the state of an [`Entry`] and
@@ -285,7 +296,7 @@ impl Filter<FilterValid> {
 
     pub fn get_attr_set(&self) -> BTreeSet<&str> {
         // Recurse through the filter getting an attribute set.
-        let mut r_set: BTreeSet<&str> = BTreeSet::new();
+        let mut r_set = BTreeSet::new();
         self.state.inner.get_attr_set(&mut r_set);
         r_set
     }
@@ -473,6 +484,7 @@ impl FilterComp {
             FC::LessThan(a, v) => FilterComp::LessThan(a.to_string(), v),
             FC::Or(v) => FilterComp::Or(v.into_iter().map(FilterComp::new).collect()),
             FC::And(v) => FilterComp::And(v.into_iter().map(FilterComp::new).collect()),
+            FC::Inclusion(v) => FilterComp::Inclusion(v.into_iter().map(FilterComp::new).collect()),
             FC::AndNot(b) => FilterComp::AndNot(Box::new(FilterComp::new(*b))),
             FC::SelfUUID => FilterComp::SelfUUID,
         }
@@ -511,6 +523,7 @@ impl FilterComp {
             }
             FilterComp::Or(vs) => vs.iter().for_each(|f| f.get_attr_set(r_set)),
             FilterComp::And(vs) => vs.iter().for_each(|f| f.get_attr_set(r_set)),
+            FilterComp::Inclusion(vs) => vs.iter().for_each(|f| f.get_attr_set(r_set)),
             FilterComp::AndNot(f) => f.get_attr_set(r_set),
             FilterComp::SelfUUID => {
                 r_set.insert("uuid");
@@ -615,6 +628,17 @@ impl FilterComp {
                 // Now put the valid filters into the Filter
                 x.map(FilterComp::And)
             }
+            FilterComp::Inclusion(filters) => {
+                if filters.is_empty() {
+                    return Err(SchemaError::EmptyFilter);
+                };
+                let x: Result<Vec<_>, _> = filters
+                    .iter()
+                    .map(|filter| filter.validate(schema))
+                    .collect();
+                // Now put the valid filters into the Filter
+                x.map(FilterComp::Inclusion)
+            }
             FilterComp::AndNot(filter) => {
                 // Just validate the inner
                 filter
@@ -859,6 +883,11 @@ impl FilterResolved {
                     .map(|v| FilterResolved::from_invalid(v, idxmeta))
                     .collect(),
             ),
+            FilterComp::Inclusion(vs) => FilterResolved::Inclusion(
+                vs.into_iter()
+                    .map(|v| FilterResolved::from_invalid(v, idxmeta))
+                    .collect(),
+            ),
             FilterComp::AndNot(f) => {
                 // TODO: pattern match box here. (AndNot(box f)).
                 // We have to clone f into our space here because pattern matching can
@@ -911,6 +940,13 @@ impl FilterResolved {
                     .collect();
                 fi.map(FilterResolved::And)
             }
+            FilterComp::Inclusion(vs) => {
+                let fi: Option<Vec<_>> = vs
+                    .into_iter()
+                    .map(|f| FilterResolved::resolve_idx(f, ev, idxmeta))
+                    .collect();
+                fi.map(FilterResolved::Inclusion)
+            }
             FilterComp::AndNot(f) => {
                 // TODO: pattern match box here. (AndNot(box f)).
                 // We have to clone f into our space here because pattern matching can
@@ -955,6 +991,13 @@ impl FilterResolved {
                     .collect();
                 fi.map(FilterResolved::And)
             }
+            FilterComp::Inclusion(vs) => {
+                let fi: Option<Vec<_>> = vs
+                    .into_iter()
+                    .map(|f| FilterResolved::resolve_no_idx(f, ev))
+                    .collect();
+                fi.map(FilterResolved::Inclusion)
+            }
             FilterComp::AndNot(f) => {
                 // TODO: pattern match box here. (AndNot(box f)).
                 // We have to clone f into our space here because pattern matching can
@@ -978,6 +1021,26 @@ impl FilterResolved {
     fn optimise(&self) -> Self {
         // Most optimisations only matter around or/and terms.
         match self {
+            FilterResolved::Inclusion(f_list) => {
+                // first, optimise all our inner elements
+                let (f_list_inc, mut f_list_new): (Vec<_>, Vec<_>) = f_list
+                    .iter()
+                    .map(|f_ref| f_ref.optimise())
+                    .partition(|f| match f {
+                        FilterResolved::Inclusion(_) => true,
+                        _ => false,
+                    });
+
+                f_list_inc.into_iter().for_each(|fc| {
+                    if let FilterResolved::Inclusion(mut l) = fc {
+                        f_list_new.append(&mut l)
+                    }
+                });
+                // finally, optimise this list by sorting.
+                f_list_new.sort_unstable();
+                f_list_new.dedup();
+                FilterResolved::Inclusion(f_list_new)
+            }
             FilterResolved::And(f_list) => {
                 // first, optimise all our inner elements
                 let (f_list_and, mut f_list_new): (Vec<_>, Vec<_>) = f_list
diff --git a/kanidmd/src/lib/macros.rs b/kanidmd/src/lib/macros.rs
index c7834255f..fe15ebbed 100644
--- a/kanidmd/src/lib/macros.rs
+++ b/kanidmd/src/lib/macros.rs
@@ -104,7 +104,7 @@ macro_rules! entry_str_to_account {
             .get_ava_single_str("name")
             .map(|s| Value::new_spn_str(s, "example.com"))
             .expect("Failed to munge spn from name!");
-        e.set_avas("spn", vec![spn]);
+        e.set_ava("spn", btreeset![spn]);
 
         let e = unsafe { e.into_sealed_committed() };
 
@@ -180,6 +180,18 @@ macro_rules! f_and {
     }};
 }
 
+#[allow(unused_macros)]
+#[macro_export]
+macro_rules! f_inc {
+    (
+        $vs:expr
+    ) => {{
+        use crate::filter::FC;
+        let s: Box<[FC]> = Box::new($vs);
+        f_inc(s.into_vec())
+    }};
+}
+
 #[allow(unused_macros)]
 #[macro_export]
 macro_rules! f_or {
@@ -203,7 +215,7 @@ macro_rules! filter {
         use crate::filter::FC;
         #[allow(unused_imports)]
         use crate::filter::{
-            f_and, f_andnot, f_eq, f_id, f_lt, f_or, f_pres, f_self, f_spn_name, f_sub,
+            f_and, f_andnot, f_eq, f_id, f_inc, f_lt, f_or, f_pres, f_self, f_spn_name, f_sub,
         };
         Filter::new_ignore_hidden($fc)
     }};
@@ -219,7 +231,9 @@ macro_rules! filter_rec {
         #[allow(unused_imports)]
         use crate::filter::FC;
         #[allow(unused_imports)]
-        use crate::filter::{f_and, f_andnot, f_eq, f_id, f_lt, f_or, f_pres, f_self, f_sub};
+        use crate::filter::{
+            f_and, f_andnot, f_eq, f_id, f_inc, f_lt, f_or, f_pres, f_self, f_sub,
+        };
         Filter::new_recycled($fc)
     }};
 }
@@ -234,7 +248,9 @@ macro_rules! filter_all {
         #[allow(unused_imports)]
         use crate::filter::FC;
         #[allow(unused_imports)]
-        use crate::filter::{f_and, f_andnot, f_eq, f_id, f_lt, f_or, f_pres, f_self, f_sub};
+        use crate::filter::{
+            f_and, f_andnot, f_eq, f_id, f_inc, f_lt, f_or, f_pres, f_self, f_sub,
+        };
         Filter::new($fc)
     }};
 }
@@ -247,7 +263,7 @@ macro_rules! filter_valid {
         $fc:expr
     ) => {{
         #[allow(unused_imports)]
-        use crate::filter::{f_and, f_andnot, f_eq, f_lt, f_or, f_pres, f_sub};
+        use crate::filter::{f_and, f_andnot, f_eq, f_inc, f_lt, f_or, f_pres, f_sub};
         use crate::filter::{Filter, FilterInvalid};
         let f: Filter<FilterInvalid> = Filter::new($fc);
         // Create a resolved filter, via the most unsafe means possible!
@@ -263,7 +279,7 @@ macro_rules! filter_resolved {
         $fc:expr
     ) => {{
         #[allow(unused_imports)]
-        use crate::filter::{f_and, f_andnot, f_eq, f_lt, f_or, f_pres, f_sub};
+        use crate::filter::{f_and, f_andnot, f_eq, f_inc, f_lt, f_or, f_pres, f_sub};
         use crate::filter::{Filter, FilterInvalid};
         let f: Filter<FilterInvalid> = Filter::new($fc);
         // Create a resolved filter, via the most unsafe means possible!
diff --git a/kanidmd/src/lib/plugins/attrunique.rs b/kanidmd/src/lib/plugins/attrunique.rs
index 188c7f88a..946b9c469 100644
--- a/kanidmd/src/lib/plugins/attrunique.rs
+++ b/kanidmd/src/lib/plugins/attrunique.rs
@@ -26,48 +26,39 @@ fn get_cand_attr_set<VALID, STATE>(
 ) -> Result<BTreeMap<PartialValue, PartialValue>, OperationError> {
     let mut cand_attr: BTreeMap<PartialValue, PartialValue> = BTreeMap::new();
 
-    for e in cand.iter() {
-        let uuid = match e.get_ava_single("uuid") {
-            Some(v) => v.to_partialvalue(),
-            None => {
-                return Err(OperationError::InvalidEntryState);
-            }
-        };
-        // Get the value and uuid
-        //for each value in the ava.
-        let mut values: Vec<PartialValue> = match e.get_ava(attr) {
-            Some(vs) => {
-                // We have values, map them.
-                vs.map(|v| v.to_partialvalue()).collect()
-            }
-            None => {
-                // No values, so empty set.
-                Vec::new()
-            }
-        };
-
-        for v in values.drain(..) {
-            match cand_attr.insert(v, uuid.clone()) {
-                // Didn't exist, move on.
-                None => {}
-                // The duplicate/rejected value moved out of the tree
-                Some(vr) => {
-                    ladmin_error!(
-                        au,
-                        "ava already exists -> {:?}: {:?} on {:?}",
-                        attr,
-                        vr,
-                        uuid
-                    );
-                    return Err(OperationError::Plugin(PluginError::AttrUnique(
-                        "ava already exists".to_string(),
-                    )));
+    cand.iter()
+        .try_for_each(|e| {
+            let uuid = match e.get_ava_single("uuid") {
+                Some(v) => v.to_partialvalue(),
+                None => {
+                    return Err(OperationError::InvalidEntryState);
                 }
-            }
-        }
-    }
-
-    Ok(cand_attr)
+            };
+            // Get the value and uuid
+            //for each value in the ava.
+            e.get_ava(attr)
+                .map(|vs| {
+                    vs.map(|v| v.to_partialvalue()).try_for_each(|v| {
+                        match cand_attr.insert(v, uuid.clone()) {
+                            None => Ok(()),
+                            Some(vr) => {
+                                ladmin_error!(
+                                    au,
+                                    "ava already exists -> {:?}: {:?} on {:?}",
+                                    attr,
+                                    vr,
+                                    uuid
+                                );
+                                Err(OperationError::Plugin(PluginError::AttrUnique(
+                                    "ava already exists".to_string(),
+                                )))
+                            }
+                        }
+                    })
+                })
+                .unwrap_or(Ok(()))
+        })
+        .map(|()| cand_attr)
 }
 
 fn enforce_unique<STATE>(
diff --git a/kanidmd/src/lib/plugins/base.rs b/kanidmd/src/lib/plugins/base.rs
index e2283e78b..b3ca26cb8 100644
--- a/kanidmd/src/lib/plugins/base.rs
+++ b/kanidmd/src/lib/plugins/base.rs
@@ -1,4 +1,5 @@
 use crate::plugins::Plugin;
+use hashbrown::HashSet;
 use std::collections::BTreeSet;
 use uuid::Uuid;
 
@@ -56,7 +57,7 @@ impl Plugin for Base {
             ltrace!(au, "Base check on entry: {:?}", entry);
 
             // First, ensure we have the 'object', class in the class set.
-            entry.add_ava("class", &CLASS_OBJECT);
+            entry.add_ava("class", CLASS_OBJECT.clone());
 
             ltrace!(au, "Object should now be in entry: {:?}", entry);
 
@@ -66,9 +67,9 @@ impl Plugin for Base {
             match entry.get_ava_set("uuid").map(|s| s.len()) {
                 None => {
                     // Generate
-                    let ava_uuid: Vec<Value> = vec![Value::new_uuid(Uuid::new_v4())];
-                    ltrace!(au, "Setting temporary UUID {:?} to entry", ava_uuid[0]);
-                    entry.set_avas("uuid", ava_uuid);
+                    let ava_uuid = btreeset![Value::new_uuid(Uuid::new_v4())];
+                    ltrace!(au, "Setting temporary UUID {:?} to entry", ava_uuid);
+                    entry.set_ava("uuid", ava_uuid);
                 }
                 Some(1) => {
                     // Do nothing
@@ -208,9 +209,6 @@ impl Plugin for Base {
         au: &mut AuditScope,
         qs: &QueryServerReadTransaction,
     ) -> Vec<Result<(), ConsistencyError>> {
-        // Verify all uuid's are unique?
-        // Probably the literally worst thing ...
-
         // Search for class = *
         let entries = match qs.internal_search(au, filter!(f_pres("class"))) {
             Ok(v) => v,
@@ -220,6 +218,8 @@ impl Plugin for Base {
             }
         };
 
+        let mut uuid_seen: HashSet<Uuid> = HashSet::with_capacity(entries.len());
+
         entries
             .iter()
             // do an exists checks on the uuid
@@ -228,20 +228,13 @@ impl Plugin for Base {
                 // will be thrown in the deserialise (possibly it will be better
                 // handled later). But it means this check only needs to validate
                 // uniqueness!
-                let uuid: &Uuid = e.get_uuid();
+                let uuid = e.get_uuid();
 
-                let filt = filter!(FC::Eq("uuid", PartialValue::new_uuid(*uuid)));
-                match qs.internal_search(au, filt) {
-                    Ok(r) => {
-                        if r.is_empty() {
-                            Err(ConsistencyError::UuidIndexCorrupt(uuid.to_string()))
-                        } else if r.len() == 1 {
-                            Ok(())
-                        } else {
-                            Err(ConsistencyError::UuidNotUnique(uuid.to_string()))
-                        }
-                    }
-                    Err(_) => Err(ConsistencyError::QueryServerSearchFailure),
+                if uuid_seen.insert(*uuid) {
+                    // Insert returns true if the item was unique.
+                    Ok(())
+                } else {
+                    Err(ConsistencyError::UuidNotUnique(uuid.to_string()))
                 }
             })
             .filter(|v| v.is_err())
diff --git a/kanidmd/src/lib/plugins/domain.rs b/kanidmd/src/lib/plugins/domain.rs
index 9064e074a..7b84afbf0 100644
--- a/kanidmd/src/lib/plugins/domain.rs
+++ b/kanidmd/src/lib/plugins/domain.rs
@@ -39,12 +39,12 @@ impl Plugin for Domain {
             {
                 // We always set this, because the DB uuid is authorative.
                 let u = Value::new_uuid(qs.get_domain_uuid());
-                e.set_avas("domain_uuid", vec![u]);
+                e.set_ava("domain_uuid", btreeset![u]);
                 ltrace!(au, "plugin_domain: Applying uuid transform");
                 // We only apply this if one isn't provided.
                 if !e.attribute_pres("domain_name") {
                     let n = Value::new_iname_s("example.com");
-                    e.set_avas("domain_name", vec![n]);
+                    e.set_ava("domain_name", btreeset![n]);
                     ltrace!(au, "plugin_domain: Applying domain_name transform");
                 }
                 ltrace!(au, "{:?}", e);
diff --git a/kanidmd/src/lib/plugins/gidnumber.rs b/kanidmd/src/lib/plugins/gidnumber.rs
index 6a881cfc6..8f41d8546 100644
--- a/kanidmd/src/lib/plugins/gidnumber.rs
+++ b/kanidmd/src/lib/plugins/gidnumber.rs
@@ -55,7 +55,7 @@ fn apply_gidnumber<T: Clone>(
 
         let gid_v = Value::new_uint32(gid);
         ladmin_info!(au, "Generated {} for {:?}", gid, u_ref);
-        e.set_avas("gidnumber", vec![gid_v]);
+        e.set_ava("gidnumber", btreeset![gid_v]);
         Ok(())
     } else if let Some(gid) = e.get_ava_single_uint32("gidnumber") {
         // If they provided us with a gid number, ensure it's in a safe range.
diff --git a/kanidmd/src/lib/plugins/memberof.rs b/kanidmd/src/lib/plugins/memberof.rs
index 4b9a6c2fe..30d9758fd 100644
--- a/kanidmd/src/lib/plugins/memberof.rs
+++ b/kanidmd/src/lib/plugins/memberof.rs
@@ -13,178 +13,195 @@
 use crate::audit::AuditScope;
 use crate::entry::{Entry, EntryCommitted, EntryInvalid, EntrySealed};
 use crate::event::{CreateEvent, DeleteEvent, ModifyEvent};
-use crate::modify::{Modify, ModifyList};
+// use crate::modify::{Modify, ModifyList};
 use crate::plugins::Plugin;
 use crate::server::QueryServerTransaction;
 use crate::server::{QueryServerReadTransaction, QueryServerWriteTransaction};
 use crate::value::{PartialValue, Value};
 use kanidm_proto::v1::{ConsistencyError, OperationError};
 
+use hashbrown::HashMap;
 use std::collections::BTreeSet;
 use uuid::Uuid;
 
 lazy_static! {
     static ref CLASS_GROUP: PartialValue = PartialValue::new_class("group");
+    static ref CLASS_MEMBEROF: Value = Value::new_class("memberof");
 }
 
 pub struct MemberOf;
 
-fn affected_uuids<'a, STATE>(
+type EntrySealedCommitted = Entry<EntrySealed, EntryCommitted>;
+type EntryInvalidCommitted = Entry<EntryInvalid, EntryCommitted>;
+type EntryTuple = (EntrySealedCommitted, EntryInvalidCommitted);
+
+fn do_memberof(
     au: &mut AuditScope,
-    changed: Vec<&'a Entry<EntrySealed, STATE>>,
-) -> Vec<&'a Uuid>
-where
-    STATE: std::fmt::Debug,
-{
-    // From the list of groups which were changed in this operation:
-    // let changed_groups: Vec<_> = changed
-    let mut affected_uuids: Vec<&Uuid> = changed
-        .into_iter()
-        .filter(|e| e.attribute_value_pres("class", &CLASS_GROUP))
-        /*
-            .collect();
-        ltrace!(au, "groups reporting change: {:?}", changed_groups);
-        // Now, build a map of all UUID's that will require updates as a result of this change
-        let mut affected_uuids: Vec<&Uuid> = changed_groups
-            .iter()
-            */
-        .filter_map(|e| {
-            // Only groups with member get collected up here.
-            e.get_ava("member")
-        })
-        // Flatten the member's to the list.
-        .flatten()
-        .filter_map(|uv| uv.to_ref_uuid())
-        .collect();
+    qs: &QueryServerWriteTransaction,
+    uuid: &Uuid,
+    tgte: &mut EntryInvalidCommitted,
+) -> Result<(), OperationError> {
+    //  search where we are member
+    let groups = qs
+        .internal_search(
+            au,
+            filter!(f_and!([
+                f_eq("class", CLASS_GROUP.clone()),
+                f_eq("member", PartialValue::new_refer(*uuid))
+            ])),
+        )
+        .map_err(|e| {
+            ladmin_error!(au, "internal search failure -> {:?}", e);
+            e
+        })?;
 
-    ltrace!(au, "uuids reporting change: {:?}", affected_uuids);
+    // Ensure we are MO capable.
+    tgte.add_ava("class", CLASS_MEMBEROF.clone());
+    // Clear the dmo + mos, we will recreate them now.
+    // This is how we handle deletes/etc.
+    tgte.pop_ava("memberof");
+    tgte.pop_ava("directmemberof");
+    // Add all the direct mo's and mos.
+    groups.iter().for_each(|g| {
+        // TODO: Change add_ava to remove this alloc/clone.
+        let dmo = Value::new_refer(*g.get_uuid());
+        tgte.add_ava("directmemberof", dmo.clone());
+        tgte.add_ava("memberof", dmo);
 
-    // IDEA: promote groups to head of the affected_uuids set!
-    //
-    // This isn't worth doing - it's only used in create/delete, it would not
-    // really make a large performance difference. Better to target improvements
-    // in the apply_memberof fn.
-    affected_uuids.sort();
-    // Remove dups
-    affected_uuids.dedup();
+        if let Some(miter) = g.get_ava("memberof") {
+            miter.for_each(|mo| {
+                tgte.add_ava("memberof", mo.clone());
+            })
+        };
+    });
 
-    affected_uuids
+    ltrace!(
+        au,
+        "Updating {:?} to be dir mo {:?}",
+        uuid,
+        tgte.get_ava_set("directmemberof")
+    );
+    ltrace!(
+        au,
+        "Updating {:?} to be mo {:?}",
+        uuid,
+        tgte.get_ava_set("memberof")
+    );
+
+    Ok(())
 }
 
+#[allow(clippy::cognitive_complexity)]
 fn apply_memberof(
     au: &mut AuditScope,
     qs: &QueryServerWriteTransaction,
-    affected_uuids: Vec<&Uuid>,
+    // TODO: Experiment with HashSet/BTreeSet here instead of vec.
+    // May require https://github.com/rust-lang/rust/issues/62924 to allow poping
+    mut group_affect: Vec<Uuid>,
 ) -> Result<(), OperationError> {
     ltrace!(au, " => entering apply_memberof");
-    ltrace!(au, "affected uuids -> {:?}", affected_uuids);
+    ltrace!(au, " => initial group_affect {:?}", group_affect);
 
-    // Apply member takes a list of changes. We then filter that to only the changed groups
-    // and using this, we determine a list of UUID's from members that will be required to
-    // re-examine their MO attributes.
+    // We can't cache groups, because we need to be continually writing
+    // and querying them. But we can cache anything we find in the process
+    // to speed up the later other_affect write op, and we can use this
+    // to avoid loading things that aren't groups.
+    // All other changed entries (mo, dmo cleared)
+    let mut other_cache: HashMap<Uuid, EntryTuple> = HashMap::with_capacity(group_affect.len() * 2);
+    while !group_affect.is_empty() {
+        group_affect.sort();
+        group_affect.dedup();
+        // Prep the write lists
+        let mut pre_candidates = Vec::with_capacity(group_affect.len());
+        let mut candidates = Vec::with_capacity(group_affect.len());
 
-    // Given the list of UUID that require changes, we attempt to trigger MO updates on groups
-    // first to stabilise the MO graph before we start triggering changes on entries.
-    //
-    // it's important to note that each change itself, especially groups, could trigger there
-    // own recursive updates until the base case - stable, no changes - is reached.
-    //
-    // That means the termination of recursion is ALWAYS to be found in the post_modify
-    // callback, as regardless of initial entry point, all subsequent MO internal operations
-    // are modifies - it is up to post_modify to break cycles!
+        // Ignore recycled/tombstones
+        let filt = filter!(FC::Or(
+            group_affect
+                .drain(0..)
+                .map(|u| f_eq("uuid", PartialValue::new_uuid(u)))
+                .collect()
+        ));
 
-    // Now work on the affected set.
+        let mut work_set = qs.internal_search_writeable(au, filt)?;
+        // Load the vecdeque with this batch.
 
-    // For each affected uuid
-    affected_uuids.into_iter().try_for_each(|a_uuid| {
-        // search where group + Eq("member": "uuid")
-        let groups = qs
-            .internal_search(
-                au,
-                filter!(f_and!([
-                    f_eq("class", CLASS_GROUP.clone()),
-                    f_eq("member", PartialValue::new_refer_r(a_uuid))
-                ])),
-            )
-            .map_err(|e| {
-                ladmin_error!(au, "internal search failure -> {:?}", e);
-                e
-            })?;
-        // get UUID of all groups + all memberof values
-        let mut dir_mo_set: Vec<Value> = groups
-            .iter()
-            .map(|g| {
-                // These are turned into reference values.
-                Value::new_refer(*g.get_uuid())
-            })
-            .collect();
+        while let Some((pre, mut tgte)) = work_set.pop() {
+            let guuid = *pre.get_uuid();
+            // load the entry from the db.
+            if !tgte.attribute_value_pres("class", &CLASS_GROUP) {
+                // It's not a group, we'll deal with you later. We should NOT
+                // have seen this UUID before, as either we are on the first
+                // iteration OR the checks belowe should have filtered it out.
+                ltrace!(au, "not a group, delaying update to -> {:?}", guuid);
+                other_cache.insert(guuid, (pre, tgte));
+                continue;
+            }
 
-        // No need to dedup this. Sorting could be of questionable
-        // value too though ...
-        dir_mo_set.sort();
-        dir_mo_set.dedup();
+            ltrace!(au, "=> processing group update -> {:?}", guuid);
 
-        let mut mo_set: Vec<Value> = groups
-            .iter()
-            .map(|g| {
-                // TODO #61: This could be more effecient
-                let mut v = vec![Value::new_refer(*g.get_uuid())];
-                if let Some(mos) = g.get_ava("memberof") {
-                    for mo in mos {
-                        // This is cloning the existing reference values
-                        v.push(mo.clone())
-                    }
-                }
-                v
-            })
-            .flatten()
-            .collect();
+            do_memberof(au, qs, &guuid, &mut tgte)?;
 
-        mo_set.sort();
-        mo_set.dedup();
+            // Did we change? Note we don't check if the class changed, only if mo changed.
+            if pre.get_ava_set("memberof") != tgte.get_ava_set("memberof")
+                || pre.get_ava_set("directmemberof") != tgte.get_ava_set("directmemberof")
+            {
+                // Yes we changed - we now must process all our members, as they need to
+                // inherit changes. Some of these members COULD be non groups, but we
+                // handle that in the dbload step.
+                ltrace!(
+                    au,
+                    "{:?} changed, flagging members as groups to change. ",
+                    guuid
+                );
+                if let Some(miter) = tgte.get_ava_as_refuuid("member") {
+                    group_affect.extend(miter.filter(|m| !other_cache.contains_key(m)));
+                };
 
-        ltrace!(au, "Updating {:?} to be dir mo {:?}", a_uuid, dir_mo_set);
-        ltrace!(au, "Updating {:?} to be mo {:?}", a_uuid, mo_set);
+                // push the entries to pre/cand
+                pre_candidates.push(pre);
+                candidates.push(tgte);
+            } else {
+                ltrace!(au, "{:?} stable", guuid);
+            }
+        }
 
-        // first add a purged memberof to remove all mo we no longer
-        // support.
-        // TODO #61: Could this be more efficient
-        // TODO #68: Could this affect replication? Or should the CL work out the
-        // true diff of the operation?
-        let mo_purge = vec![
-            Modify::Present("class".to_string(), Value::new_class("memberof")),
-            Modify::Purged("memberof".to_string()),
-            Modify::Purged("directmemberof".to_string()),
-        ];
+        debug_assert!(pre_candidates.len() == candidates.len());
+        // Write this stripe if populated.
+        if !pre_candidates.is_empty() {
+            qs.internal_batch_modify(au, pre_candidates, candidates)
+                .map_err(|e| {
+                    ladmin_error!(au, "Failed to commit memberof group set {:?}", e);
+                    e
+                })?;
+        }
+        // Next loop!
+    }
 
-        // create modify present memberof all uuids
-        let mod_set: Vec<_> = mo_purge
-            .into_iter()
-            .chain(
-                mo_set
-                    .into_iter()
-                    .map(|mo_uuid| Modify::Present("memberof".to_string(), mo_uuid)),
-            )
-            .chain(
-                dir_mo_set
-                    .into_iter()
-                    .map(|mo_uuid| Modify::Present("directmemberof".to_string(), mo_uuid)),
-            )
-            .collect();
+    // ALL GROUP MOS + DMOS ARE NOW STABLE. We can load these into other items directly.
+    let mut pre_candidates = Vec::with_capacity(other_cache.len());
+    let mut candidates = Vec::with_capacity(other_cache.len());
 
-        // apply to affected uuid
-        let modlist = ModifyList::new_list(mod_set);
+    other_cache
+        .into_iter()
+        .try_for_each(|(auuid, (pre, mut tgte))| {
+            ltrace!(au, "=> processing affected uuid {:?}", auuid);
+            debug_assert!(!tgte.attribute_value_pres("class", &CLASS_GROUP));
+            do_memberof(au, qs, &auuid, &mut tgte)?;
+            // Only write if a change occured.
+            if pre.get_ava_set("memberof") != tgte.get_ava_set("memberof")
+                || pre.get_ava_set("directmemberof") != tgte.get_ava_set("directmemberof")
+            {
+                pre_candidates.push(pre);
+                candidates.push(tgte);
+            }
+            Ok(())
+        })?;
 
-        qs.internal_modify(
-            au,
-            filter!(f_eq("uuid", PartialValue::new_uuid(*a_uuid))),
-            modlist,
-        )
-        .map_err(|e| {
-            ladmin_error!(au, "Internal modification failure -> {:?}", e);
-            e
-        })
-    })
+    // Turn the other_cache into a write set.
+    // Write the batch out in a single stripe.
+    qs.internal_batch_modify(au, pre_candidates, candidates)
+    // Done! 🎉
 }
 
 impl Plugin for MemberOf {
@@ -192,20 +209,31 @@ impl Plugin for MemberOf {
         "memberof"
     }
 
-    // TODO #61: We could make this more effecient by limiting change detection to ONLY member/memberof
-    // attrs rather than any attrs.
-
     fn post_create(
         au: &mut AuditScope,
         qs: &QueryServerWriteTransaction,
         cand: &[Entry<EntrySealed, EntryCommitted>],
         _ce: &CreateEvent,
     ) -> Result<(), OperationError> {
-        //
-        // Trigger apply_memberof on all because they changed.
-        let cand_refs: Vec<&Entry<_, _>> = cand.iter().map(|e| e).collect();
-        let uuids = affected_uuids(au, cand_refs);
-        apply_memberof(au, qs, uuids)
+        let group_affect = cand
+            .iter()
+            .map(|e| e.get_uuid())
+            .chain(
+                cand.iter()
+                    .filter_map(|e| {
+                        // Is it a group?
+                        if e.attribute_value_pres("class", &CLASS_GROUP) {
+                            e.get_ava_as_refuuid("member")
+                        } else {
+                            None
+                        }
+                    })
+                    .flatten(),
+            )
+            .copied()
+            .collect();
+
+        apply_memberof(au, qs, group_affect)
     }
 
     fn post_modify(
@@ -215,40 +243,37 @@ impl Plugin for MemberOf {
         cand: &[Entry<EntrySealed, EntryCommitted>],
         _me: &ModifyEvent,
     ) -> Result<(), OperationError> {
-        // The condition here is critical - ONLY trigger on entries where changes occur!
-        let mut changed: Vec<&Uuid> = pre_cand
+        // TODO: Limit this to when it's a class, member, mo, dmo change instead.
+        let group_affect = cand
             .iter()
-            .zip(cand.iter())
-            .filter(|(pre, post)| {
-                // This is the base case to break cycles in recursion!
-                (
-                        // If it was a group, or will become a group.
-                        post.attribute_value_pres("class", &CLASS_GROUP)
-                            || pre.attribute_value_pres("class", &CLASS_GROUP)
-                    )
-                    // And the group has changed ...
-                    && pre != post
-                // Then memberof should be updated!
-            })
-            // Flatten the pre-post tuples. We no longer care if it was
-            // pre-post
-            .flat_map(|(pre, post)| vec![pre, post])
-            .inspect(|e| {
-                ltrace!(au, "group reporting change: {:?}", e);
-            })
-            .filter_map(|e| {
-                // Only groups with member get collected up here.
-                e.get_ava_as_refuuid("member")
-            })
-            // Flatten the uuid reference lists.
-            .flatten()
+            .map(|post| post.get_uuid())
+            .chain(
+                pre_cand
+                    .iter()
+                    .filter_map(|pre| {
+                        if pre.attribute_value_pres("class", &CLASS_GROUP) {
+                            pre.get_ava_as_refuuid("member")
+                        } else {
+                            None
+                        }
+                    })
+                    .flatten(),
+            )
+            .chain(
+                cand.iter()
+                    .filter_map(|post| {
+                        if post.attribute_value_pres("class", &CLASS_GROUP) {
+                            post.get_ava_as_refuuid("member")
+                        } else {
+                            None
+                        }
+                    })
+                    .flatten(),
+            )
+            .copied()
             .collect();
 
-        // Now tidy them up to reduce excesse searches/work.
-        changed.sort();
-        changed.dedup();
-
-        apply_memberof(au, qs, changed)
+        apply_memberof(au, qs, group_affect)
     }
 
     fn pre_delete(
@@ -265,15 +290,8 @@ impl Plugin for MemberOf {
         // As a result, on restore, the graph of where it was a member
         // would have to be rebuilt.
         //
-        // AN interesting possibility could be NOT to purge MO on delete
-        // and use that to rebuild the forward graph of member -> item, but
-        // due to the nature of MO, we do not know the difference between
-        // direct and indirect membership, meaning we would be safer
-        // to not do this.
-
         // NOTE: DO NOT purge directmemberof - we use that to restore memberships
         // in recycle revive!
-
         cand.iter_mut().for_each(|e| e.purge_ava("memberof"));
         Ok(())
     }
@@ -284,11 +302,23 @@ impl Plugin for MemberOf {
         cand: &[Entry<EntrySealed, EntryCommitted>],
         _ce: &DeleteEvent,
     ) -> Result<(), OperationError> {
-        //
-        // Trigger apply_memberof on all - because they all changed.
-        let cand_refs: Vec<&Entry<_, _>> = cand.iter().map(|e| e).collect();
-        let uuids = affected_uuids(au, cand_refs);
-        apply_memberof(au, qs, uuids)
+        // Similar condition to create - we only trigger updates on groups's members,
+        // so that they can find they are no longer a mo of what was deleted.
+        let group_affect = cand
+            .iter()
+            .filter_map(|e| {
+                // Is it a group?
+                if e.attribute_value_pres("class", &CLASS_GROUP) {
+                    e.get_ava_as_refuuid("member")
+                } else {
+                    None
+                }
+            })
+            .flatten()
+            .copied()
+            .collect();
+
+        apply_memberof(au, qs, group_affect)
     }
 
     fn verify(
@@ -309,9 +339,6 @@ impl Plugin for MemberOf {
 
         // for each entry in the DB (live).
         for e in all_cand {
-            // create new map
-            // let mo_set: BTreeMap<String, ()> = BTreeMap::new();
-            // searcch direct memberships of live groups.
             let filt_in = filter!(f_and!([
                 f_eq("class", PartialValue::new_class("group")),
                 f_eq("member", PartialValue::new_refer(*e.get_uuid()))
@@ -326,8 +353,16 @@ impl Plugin for MemberOf {
             };
             // for all direct -> add uuid to map
 
-            let d_groups_set: BTreeSet<&Uuid> =
-                direct_memberof.iter().map(|e| e.get_uuid()).collect();
+            let d_groups_set: BTreeSet<_> = direct_memberof
+                .iter()
+                .map(|e| Value::new_refer(*e.get_uuid()))
+                .collect();
+
+            let d_groups_set = if d_groups_set.is_empty() {
+                None
+            } else {
+                Some(d_groups_set)
+            };
 
             ltrace!(
                 au,
@@ -336,40 +371,32 @@ impl Plugin for MemberOf {
                 d_groups_set
             );
 
-            let dmos: Vec<&Uuid> = match e.get_ava_as_refuuid("directmemberof") {
-                // Avoid a reference issue to return empty set
-                Some(dmos) => dmos.collect(),
-                // No memberof, return empty set.
-                None => Vec::new(),
-            };
-
-            ltrace!(au, "Direct Member Of Set {:?} -> {:?}", e.get_uuid(), dmos);
-
-            if dmos.len() != direct_memberof.len() {
-                ladmin_error!(
-                    au,
-                    "MemberOfInvalid directmemberof set and DMO search set differ in size: {}",
-                    e
-                );
-                r.push(Err(ConsistencyError::MemberOfInvalid(e.get_id())));
-                debug_assert!(false);
-                // Next entry
-                continue;
-            };
-
-            for mo_uuid in dmos {
-                if !d_groups_set.contains(mo_uuid) {
+            match (e.get_ava_set("directmemberof"), d_groups_set) {
+                (Some(edmos), Some(dmos)) => {
+                    let diff: Vec<_> = dmos.symmetric_difference(edmos).collect();
+                    if !diff.is_empty() {
+                        ladmin_error!(
+                            au,
+                            "MemberOfInvalid: Entry {}, DMO has inconsistencies -> {:?}",
+                            e,
+                            diff
+                        );
+                        r.push(Err(ConsistencyError::MemberOfInvalid(e.get_id())));
+                    }
+                }
+                (None, None) => {
+                    // Ok
+                }
+                _ => {
                     ladmin_error!(
                         au,
-                        "MemberOfInvalid: Entry {}, MO {:?} not in direct groups",
-                        e,
-                        mo_uuid
+                        "MemberOfInvalid directmemberof set and DMO search set differ in size: {}",
+                        e.get_uuid()
                     );
                     r.push(Err(ConsistencyError::MemberOfInvalid(e.get_id())));
-                    // Next entry
-                    continue;
                 }
             }
+
             // Could check all dmos in mos?
 
             /* To check nested! */
@@ -507,7 +534,7 @@ mod tests {
 
         let eb: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EB);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
 
         let preload = Vec::new();
         let create = vec![ea, eb];
@@ -537,8 +564,8 @@ mod tests {
 
         let ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
 
         let preload = Vec::new();
         let create = vec![ea, eb, ec];
@@ -588,9 +615,9 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = Vec::new();
         let create = vec![ea, eb, ec];
@@ -642,13 +669,13 @@ mod tests {
 
         let mut ed: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(ED);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
 
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("member", &Value::new_refer_s(&UUID_D).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_D).unwrap());
 
-        ed.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
+        ed.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = Vec::new();
         let create = vec![ea, eb, ec, ed];
@@ -745,7 +772,7 @@ mod tests {
 
         let ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -798,7 +825,7 @@ mod tests {
 
         let ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -853,8 +880,8 @@ mod tests {
 
         let ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -915,9 +942,9 @@ mod tests {
 
         let ed: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(ED);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("member", &Value::new_refer_s(&UUID_D).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_D).unwrap());
 
         let preload = vec![ea, eb, ec, ed];
         run_modify_test!(
@@ -987,8 +1014,8 @@ mod tests {
 
         let mut eb: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EB);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = vec![ea, eb];
         run_modify_test!(
@@ -1023,10 +1050,10 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -1079,11 +1106,11 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -1137,20 +1164,20 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_modify_test!(
@@ -1212,30 +1239,30 @@ mod tests {
 
         let mut ed: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(ED);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("member", &Value::new_refer_s(&UUID_D).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_D).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        ed.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ed.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = vec![ea, eb, ec, ed];
         run_modify_test!(
@@ -1306,8 +1333,8 @@ mod tests {
 
         let mut eb: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EB);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
         let preload = vec![ea, eb];
         run_delete_test!(
@@ -1336,12 +1363,12 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_delete_test!(
@@ -1380,12 +1407,12 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_delete_test!(
@@ -1425,20 +1452,20 @@ mod tests {
 
         let mut ec: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(EC);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
 
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
 
         let preload = vec![ea, eb, ec];
         run_delete_test!(
@@ -1481,30 +1508,30 @@ mod tests {
 
         let mut ed: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(ED);
 
-        ea.add_ava("member", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ea.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
+        ea.add_ava("member", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ea.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
 
-        eb.add_ava("member", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        eb.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
+        eb.add_ava("member", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        eb.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
 
-        ec.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("member", &Value::new_refer_s(&UUID_D).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ec.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("member", Value::new_refer_s(&UUID_D).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ec.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
 
-        ed.add_ava("member", &Value::new_refer_s(&UUID_A).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_A).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_B).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_C).unwrap());
-        ed.add_ava("memberof", &Value::new_refer_s(&UUID_D).unwrap());
+        ed.add_ava("member", Value::new_refer_s(&UUID_A).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_A).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_B).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_C).unwrap());
+        ed.add_ava("memberof", Value::new_refer_s(&UUID_D).unwrap());
 
         let preload = vec![ea, eb, ec, ed];
         run_delete_test!(
diff --git a/kanidmd/src/lib/plugins/password_import.rs b/kanidmd/src/lib/plugins/password_import.rs
index 08c61480f..5515b2db2 100644
--- a/kanidmd/src/lib/plugins/password_import.rs
+++ b/kanidmd/src/lib/plugins/password_import.rs
@@ -57,8 +57,8 @@ impl Plugin for PasswordImport {
                     None => {
                         // just set it then!
                         let c = Credential::new_from_password(pw);
-                        e.set_avas("primary_credential",
-                            vec![Value::new_credential("primary", c)]);
+                        e.set_ava("primary_credential",
+                            btreeset![Value::new_credential("primary", c)]);
                         Ok(())
                     }
                 }
@@ -105,18 +105,18 @@ impl Plugin for PasswordImport {
                 Some(c) => {
                     // This is the major diff to create, we can update in place!
                     let c = c.update_password(pw);
-                    e.set_avas(
+                    e.set_ava(
                         "primary_credential",
-                        vec![Value::new_credential("primary", c)],
+                        btreeset![Value::new_credential("primary", c)],
                     );
                     Ok(())
                 }
                 None => {
                     // just set it then!
                     let c = Credential::new_from_password(pw);
-                    e.set_avas(
+                    e.set_ava(
                         "primary_credential",
-                        vec![Value::new_credential("primary", c)],
+                        btreeset![Value::new_credential("primary", c)],
                     );
                     Ok(())
                 }
@@ -207,7 +207,7 @@ mod tests {
         );
 
         let c = Credential::new_password_only("password");
-        ea.add_ava("primary_credential", &Value::new_credential("primary", c));
+        ea.add_ava("primary_credential", Value::new_credential("primary", c));
 
         let preload = vec![ea];
 
@@ -241,7 +241,7 @@ mod tests {
 
         let totp = TOTP::generate_secure("test_totp".to_string(), TOTP_DEFAULT_STEP);
         let c = Credential::new_password_only("password").update_totp(totp);
-        ea.add_ava("primary_credential", &Value::new_credential("primary", c));
+        ea.add_ava("primary_credential", Value::new_credential("primary", c));
 
         let preload = vec![ea];
 
diff --git a/kanidmd/src/lib/plugins/protected.rs b/kanidmd/src/lib/plugins/protected.rs
index f306af84c..00140b6f3 100644
--- a/kanidmd/src/lib/plugins/protected.rs
+++ b/kanidmd/src/lib/plugins/protected.rs
@@ -8,8 +8,8 @@ use crate::event::{CreateEvent, DeleteEvent, ModifyEvent};
 use crate::modify::Modify;
 use crate::server::QueryServerWriteTransaction;
 use crate::value::{PartialValue, Value};
+use hashbrown::HashSet;
 use kanidm_proto::v1::OperationError;
-use std::collections::HashSet;
 
 pub struct Protected {}
 
@@ -19,7 +19,7 @@ pub struct Protected {}
 
 lazy_static! {
     static ref ALLOWED_ATTRS: HashSet<&'static str> = {
-        let mut m = HashSet::new();
+        let mut m = HashSet::with_capacity(8);
         // Allow modification of some schema class types to allow local extension
         // of schema types.
         m.insert("must");
diff --git a/kanidmd/src/lib/plugins/refint.rs b/kanidmd/src/lib/plugins/refint.rs
index bd6ebc5e3..0abac9abe 100644
--- a/kanidmd/src/lib/plugins/refint.rs
+++ b/kanidmd/src/lib/plugins/refint.rs
@@ -9,12 +9,14 @@
 // when that is written, as they *both* manipulate and alter entry reference
 // data, so we should be careful not to step on each other.
 
-use std::collections::HashSet as Set;
+use hashbrown::HashSet as Set;
+use std::collections::BTreeSet;
 
 use crate::audit::AuditScope;
 use crate::entry::{Entry, EntryCommitted, EntrySealed};
 use crate::event::{CreateEvent, DeleteEvent, ModifyEvent};
-use crate::modify::{Modify, ModifyInvalid, ModifyList};
+use crate::filter::f_eq;
+use crate::modify::Modify;
 use crate::plugins::Plugin;
 use crate::schema::SchemaTransaction;
 use crate::server::QueryServerTransaction;
@@ -28,32 +30,52 @@ use uuid::Uuid;
 pub struct ReferentialIntegrity;
 
 impl ReferentialIntegrity {
-    fn check_uuid_exists(
+    fn check_uuids_exist<'a, I>(
         au: &mut AuditScope,
         qs: &QueryServerWriteTransaction,
-        rtype: &str,
-        uuid_value: &Value,
-    ) -> Result<(), OperationError> {
-        let uuid = uuid_value.to_ref_uuid().ok_or_else(|| {
-            ladmin_error!(au, "uuid value could not convert to reference uuid");
-            OperationError::InvalidAttribute("uuid could not become reference value".to_string())
-        })?;
-        // NOTE: This only checks LIVE entries (not using filter_all)
-        let filt_in = filter!(f_eq("uuid", PartialValue::new_uuid(*uuid)));
+        ref_iter: I,
+    ) -> Result<(), OperationError>
+    where
+        I: Iterator<Item = &'a Value>,
+    {
+        let inner: Result<Vec<_>, _> = ref_iter
+            .map(|uuid_value| {
+                uuid_value
+                    .to_ref_uuid()
+                    .map(|uuid| f_eq("uuid", PartialValue::new_uuid(*uuid)))
+                    .ok_or_else(|| {
+                        ladmin_error!(au, "ref value could not convert to reference uuid");
+                        OperationError::InvalidAttribute(
+                            "uuid could not become reference value".to_string(),
+                        )
+                    })
+            })
+            .collect();
+
+        let inner = inner?;
+
+        if inner.is_empty() {
+            // There is nothing to check! Move on.
+            ladmin_info!(au, "no reference types modified, skipping check");
+            return Ok(());
+        }
+
+        // F_inc(lusion). All items of inner must be 1 or more, or the filter
+        // will fail. This will return the union of the inclusion after the
+        // operationn.
+        let filt_in = filter!(f_inc(inner));
         let b = qs.internal_exists(au, filt_in).map_err(|e| {
             ladmin_error!(au, "internal exists failure -> {:?}", e);
             e
         })?;
 
-        // Is the reference in the result set?
+        // Is the existance of all id's confirmed?
         if b {
             Ok(())
         } else {
             ladmin_error!(
                 au,
-                "{:?}:{:?} UUID reference not found in database",
-                rtype,
-                uuid
+                "UUID reference set size differs from query result size <fast path, no uuid info available>"
             );
             Err(OperationError::Plugin(PluginError::ReferentialIntegrity(
                 "Uuid referenced not found in database".to_string(),
@@ -90,20 +112,18 @@ impl Plugin for ReferentialIntegrity {
         let schema = qs.get_schema();
         let ref_types = schema.get_reference_types();
 
-        // For all cands
-        for c in cand {
-            // For all reference in each cand.
-            for rtype in ref_types.values() {
-                // If the attribute is present
-                if let Some(vs) = c.get_ava(&rtype.name) {
-                    // For each value in the set.
-                    for v in vs {
-                        Self::check_uuid_exists(au, qs, &rtype.name, v)?
-                    }
-                }
-            }
-        }
-        Ok(())
+        // Fast Path
+        let i = cand
+            .iter()
+            .map(|c| {
+                ref_types
+                    .values()
+                    .filter_map(move |rtype| c.get_ava(&rtype.name))
+            })
+            .flatten()
+            .flatten();
+
+        Self::check_uuids_exist(au, qs, i)
     }
 
     fn post_modify(
@@ -116,17 +136,19 @@ impl Plugin for ReferentialIntegrity {
         let schema = qs.get_schema();
         let ref_types = schema.get_reference_types();
 
-        // For all mods
-        for modify in me.modlist.into_iter() {
-            // If the mod affects a reference type and being ADDED.
+        let i = me.modlist.into_iter().filter_map(|modify| {
             if let Modify::Present(a, v) = &modify {
-                if let Some(a_type) = ref_types.get(a) {
-                    // So it is a reference type, now check it.
-                    Self::check_uuid_exists(au, qs, &a_type.name, v)?
+                if ref_types.get(a).is_some() {
+                    Some(v)
+                } else {
+                    None
                 }
+            } else {
+                None
             }
-        }
-        Ok(())
+        });
+
+        Self::check_uuids_exist(au, qs, i)
     }
 
     fn post_delete(
@@ -163,27 +185,24 @@ impl Plugin for ReferentialIntegrity {
 
         ltrace!(au, "refint post_delete filter {:?}", filt);
 
-        // Create a modlist:
-        //    In each, create a "removed" for each attr:uuid pair
-        let modlist: ModifyList<ModifyInvalid> = ModifyList::new_list(
-            // uuids
-            // .iter()
-            cand.iter()
-                .map(|e| e.get_uuid())
-                .map(|u| {
-                    ref_types.values().map(move |r_type| {
-                        Modify::Removed(r_type.name.clone(), PartialValue::new_refer(*u))
-                    })
-                })
-                .flatten()
-                .collect(),
-        );
+        let removed_ids: BTreeSet<_> = cand
+            .iter()
+            .map(|e| PartialValue::new_refer(*e.get_uuid()))
+            .collect();
 
-        ltrace!(au, "refint post_delete modlist {:?}", modlist);
+        let work_set = qs.internal_search_writeable(au, filt)?;
 
-        // Do an internal modify to apply the modlist and filter.
+        let (pre_candidates, candidates) = work_set
+            .into_iter()
+            .map(|(pre, mut post)| {
+                ref_types
+                    .values()
+                    .for_each(|attr| post.remove_avas(attr.name.as_str(), &removed_ids));
+                (pre, post)
+            })
+            .unzip();
 
-        qs.internal_modify(au, filt, modlist)
+        qs.internal_batch_modify(au, pre_candidates, candidates)
     }
 
     fn verify(
@@ -239,6 +258,7 @@ impl Plugin for ReferentialIntegrity {
 mod tests {
     // #[macro_use]
     // use crate::plugins::Plugin;
+    use crate::constants::UUID_DOES_NOT_EXIST;
     use crate::entry::{Entry, EntryInit, EntryNew};
     use crate::modify::{Modify, ModifyList};
     use crate::server::{QueryServerTransaction, QueryServerWriteTransaction};
@@ -422,6 +442,51 @@ mod tests {
         );
     }
 
+    // Check that even when SOME references exist, so long as one does not,
+    // we fail.
+    #[test]
+    fn test_modify_uuid_reference_partial_not_exist() {
+        let ea: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(
+            r#"{
+            "attrs": {
+                "class": ["group"],
+                "name": ["testgroup_a"],
+                "description": ["testgroup"],
+                "uuid": ["d2b496bd-8493-47b7-8142-f568b5cf47ee"]
+            }
+        }"#,
+        );
+
+        let eb: Entry<EntryInit, EntryNew> = Entry::unsafe_from_entry_str(
+            r#"{
+            "attrs": {
+                "class": ["group"],
+                "name": ["testgroup_b"],
+                "description": ["testgroup"]
+            }
+        }"#,
+        );
+
+        let preload = vec![ea, eb];
+
+        run_modify_test!(
+            Err(OperationError::Plugin(PluginError::ReferentialIntegrity(
+                "Uuid referenced not found in database".to_string()
+            ))),
+            preload,
+            filter!(f_eq("name", PartialValue::new_iname("testgroup_b"))),
+            ModifyList::new_list(vec![
+                Modify::Present(
+                    "member".to_string(),
+                    Value::new_refer_s("d2b496bd-8493-47b7-8142-f568b5cf47ee").unwrap()
+                ),
+                Modify::Present("member".to_string(), Value::new_refer(*UUID_DOES_NOT_EXIST)),
+            ]),
+            None,
+            |_, _| {}
+        );
+    }
+
     // Modify removes the reference to an entry
     #[test]
     fn test_modify_remove_referee() {
diff --git a/kanidmd/src/lib/plugins/spn.rs b/kanidmd/src/lib/plugins/spn.rs
index ff3200317..faf503bf3 100644
--- a/kanidmd/src/lib/plugins/spn.rs
+++ b/kanidmd/src/lib/plugins/spn.rs
@@ -100,7 +100,7 @@ impl Plugin for Spn {
                         e
                     })?;
                 ltrace!(au, "plugin_spn: set spn to {:?}", spn);
-                e.set_avas("spn", vec![spn]);
+                e.set_ava("spn", btreeset![spn]);
             }
         }
         Ok(())
@@ -141,7 +141,7 @@ impl Plugin for Spn {
                         e
                     })?;
                 ltrace!(au, "plugin_spn: set spn to {:?}", spn);
-                e.set_avas("spn", vec![spn]);
+                e.set_ava("spn", btreeset![spn]);
             }
         }
         Ok(())
@@ -255,8 +255,6 @@ impl Plugin for Spn {
                         );
                         debug_assert!(false);
                         r.push(Err(ConsistencyError::InvalidSPN(e.get_id())))
-                    } else {
-                        ltrace!(au, "spn is ok! 👍");
                     }
                 }
                 None => {
diff --git a/kanidmd/src/lib/repl/cid.rs b/kanidmd/src/lib/repl/cid.rs
index 3e69161d1..faa5aa34d 100644
--- a/kanidmd/src/lib/repl/cid.rs
+++ b/kanidmd/src/lib/repl/cid.rs
@@ -2,7 +2,7 @@ use kanidm_proto::v1::OperationError;
 use std::time::Duration;
 use uuid::Uuid;
 
-#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Eq, PartialOrd, Ord)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Eq, PartialOrd, Ord, Hash)]
 pub struct Cid {
     // Mental note: Derive ord always checks in order of struct fields.
     pub ts: Duration,
diff --git a/kanidmd/src/lib/schema.rs b/kanidmd/src/lib/schema.rs
index 64ebe0e40..25bd28c68 100644
--- a/kanidmd/src/lib/schema.rs
+++ b/kanidmd/src/lib/schema.rs
@@ -23,10 +23,10 @@ use crate::entry::{Entry, EntryCommitted, EntryInit, EntryNew, EntrySealed};
 use crate::value::{IndexType, PartialValue, SyntaxType, Value};
 use kanidm_proto::v1::{ConsistencyError, OperationError, SchemaError};
 
+use hashbrown::HashMap;
+use hashbrown::HashSet;
 use std::borrow::Borrow;
 use std::collections::BTreeSet;
-use std::collections::HashMap;
-use std::collections::HashSet;
 use uuid::Uuid;
 
 use concread::cowcell::*;
@@ -1448,10 +1448,10 @@ impl SchemaTransaction for SchemaReadTransaction {
 impl Schema {
     pub fn new(audit: &mut AuditScope) -> Result<Self, OperationError> {
         let s = Schema {
-            classes: CowCell::new(HashMap::new()),
-            attributes: CowCell::new(HashMap::new()),
+            classes: CowCell::new(HashMap::with_capacity(128)),
+            attributes: CowCell::new(HashMap::with_capacity(128)),
             unique_cache: CowCell::new(Vec::new()),
-            ref_cache: CowCell::new(HashMap::new()),
+            ref_cache: CowCell::new(HashMap::with_capacity(64)),
         };
         let mut sw = s.write();
         let r1 = sw.generate_in_memory(audit);
diff --git a/kanidmd/src/lib/server.rs b/kanidmd/src/lib/server.rs
index 08352f6c4..e3c42f9e4 100644
--- a/kanidmd/src/lib/server.rs
+++ b/kanidmd/src/lib/server.rs
@@ -4,8 +4,9 @@
 // This is really only used for long lived, high level types that need clone
 // that otherwise can't be cloned. Think Mutex.
 // use actix::prelude::*;
+use hashbrown::HashMap;
 use std::cell::Cell;
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::BTreeSet;
 use std::sync::Arc;
 use std::time::Duration;
 use uuid::Uuid;
@@ -38,6 +39,10 @@ use crate::schema::{
 use crate::value::{PartialValue, SyntaxType, Value};
 use kanidm_proto::v1::{ConsistencyError, OperationError, SchemaError};
 
+type EntrySealedCommitted = Entry<EntrySealed, EntryCommitted>;
+type EntryInvalidCommitted = Entry<EntryInvalid, EntryCommitted>;
+type EntryTuple = (EntrySealedCommitted, EntryInvalidCommitted);
+
 lazy_static! {
     static ref PVCLASS_ATTRIBUTETYPE: PartialValue = PartialValue::new_class("attributetype");
     static ref PVCLASS_CLASSTYPE: PartialValue = PartialValue::new_class("classtype");
@@ -1225,7 +1230,8 @@ impl<'a> QueryServerWriteTransaction<'a> {
             let revive_cands =
                 self.impersonate_search_valid(au, re.filter.clone(), re.filter.clone(), &re.event)?;
 
-            let mut dm_mods: BTreeMap<Uuid, ModifyList<ModifyInvalid>> = BTreeMap::new();
+            let mut dm_mods: HashMap<Uuid, ModifyList<ModifyInvalid>> =
+                HashMap::with_capacity(revive_cands.len());
 
             revive_cands.into_iter().for_each(|e| {
                 // Get this entries uuid.
@@ -1448,6 +1454,127 @@ impl<'a> QueryServerWriteTransaction<'a> {
         })
     }
 
+    /// Used in conjunction with internal_batch_modify, to get a pre/post
+    /// pair, where post is pre-configured with metadata to allow
+    /// modificiation before submit back to internal_batch_modify
+    pub(crate) fn internal_search_writeable(
+        &self,
+        audit: &mut AuditScope,
+        filter: Filter<FilterInvalid>,
+    ) -> Result<Vec<EntryTuple>, OperationError> {
+        lperf_segment!(audit, "server::internal_search_writeable", || {
+            let f_valid = filter
+                .validate(self.get_schema())
+                .map_err(OperationError::SchemaViolation)?;
+            let se = SearchEvent::new_internal(f_valid);
+            self.search(audit, &se).map(|vs| {
+                vs.into_iter()
+                    .map(|e| {
+                        let writeable = e.clone().invalidate(self.cid.clone());
+                        (e, writeable)
+                    })
+                    .collect()
+            })
+        })
+    }
+
+    /// Allows writing batches of modified entries without going through
+    /// the modlist path. This allows more effecient batch transformations
+    /// such as memberof, but at the expense that YOU must guarantee you
+    /// uphold all other plugin and state rules that are important. You
+    /// probably want modify instead.
+    pub(crate) fn internal_batch_modify(
+        &self,
+        au: &mut AuditScope,
+        pre_candidates: Vec<Entry<EntrySealed, EntryCommitted>>,
+        candidates: Vec<Entry<EntryInvalid, EntryCommitted>>,
+    ) -> Result<(), OperationError> {
+        lperf_segment!(au, "server::internal_batch_modify", || {
+            lsecurity!(au, "modify initiator: -> internal batch modify");
+
+            if pre_candidates.is_empty() && candidates.is_empty() {
+                // No action needed.
+                return Ok(());
+            }
+
+            if pre_candidates.len() != candidates.len() {
+                ladmin_error!(au, "internal_batch_modify - cand lengths differ");
+                return Err(OperationError::InvalidRequestState);
+            }
+
+            let res: Result<Vec<Entry<EntrySealed, EntryCommitted>>, OperationError> = candidates
+                .into_iter()
+                .map(|e| {
+                    e.validate(&self.schema)
+                        .map_err(|e| {
+                            ladmin_error!(au, "Schema Violation {:?}", e);
+                            OperationError::SchemaViolation(e)
+                        })
+                        .map(|e| e.seal())
+                })
+                .collect();
+
+            let norm_cand: Vec<Entry<_, _>> = res?;
+
+            if cfg!(debug_assertions) {
+                pre_candidates
+                    .iter()
+                    .zip(norm_cand.iter())
+                    .try_for_each(|(pre, post)| {
+                        if pre.get_uuid() == post.get_uuid() {
+                            Ok(())
+                        } else {
+                            ladmin_error!(au, "modify - cand sets not correctly aligned");
+                            Err(OperationError::InvalidRequestState)
+                        }
+                    })?;
+            }
+
+            // Backend Modify
+            self.be_txn
+                .modify(au, &pre_candidates, &norm_cand)
+                .map_err(|e| {
+                    ladmin_error!(au, "Modify operation failed (backend), {:?}", e);
+                    e
+                })?;
+
+            let _ =
+                self.changed_schema
+                    .replace(norm_cand.iter().chain(pre_candidates.iter()).fold(
+                        false,
+                        |acc, e| {
+                            if acc {
+                                acc
+                            } else {
+                                e.attribute_value_pres("class", &PVCLASS_CLASSTYPE)
+                                    || e.attribute_value_pres("class", &PVCLASS_ATTRIBUTETYPE)
+                            }
+                        },
+                    ));
+            let _ =
+                self.changed_acp
+                    .replace(norm_cand.iter().chain(pre_candidates.iter()).fold(
+                        false,
+                        |acc, e| {
+                            if acc {
+                                acc
+                            } else {
+                                e.attribute_value_pres("class", &PVCLASS_ACP)
+                            }
+                        },
+                    ));
+            ltrace!(
+                au,
+                "Schema reload: {:?}, ACP reload: {:?}",
+                self.changed_schema,
+                self.changed_acp
+            );
+
+            ltrace!(au, "Modify operation success");
+            Ok(())
+        })
+    }
+
     /// Migrate 2 to 3 changes the name, domain_name types from iutf8 to iname.
     pub fn migrate_2_to_3(&self, au: &mut AuditScope) -> Result<(), OperationError> {
         lperf_segment!(au, "server::migrate_2_to_3", || {
@@ -3423,9 +3550,9 @@ mod tests {
             }
             }"#,
         );
-        e1.add_ava("uuid", &Value::new_uuids(uuid).unwrap());
-        e1.add_ava("name", &Value::new_iname_s(name));
-        e1.add_ava("displayname", &Value::new_utf8s(name));
+        e1.add_ava("uuid", Value::new_uuids(uuid).unwrap());
+        e1.add_ava("name", Value::new_iname_s(name));
+        e1.add_ava("displayname", Value::new_utf8s(name));
         e1
     }
 
@@ -3438,11 +3565,11 @@ mod tests {
             }
             }"#,
         );
-        e1.add_ava("name", &Value::new_iname_s(name));
-        e1.add_ava("uuid", &Value::new_uuids(uuid).unwrap());
+        e1.add_ava("name", Value::new_iname_s(name));
+        e1.add_ava("uuid", Value::new_uuids(uuid).unwrap());
         members
             .iter()
-            .for_each(|m| e1.add_ava("member", &Value::new_refer_s(m).unwrap()));
+            .for_each(|m| e1.add_ava("member", Value::new_refer_s(m).unwrap()));
         e1
     }
 
diff --git a/kanidmd/src/lib/value.rs b/kanidmd/src/lib/value.rs
index ede075a16..66869e9e2 100644
--- a/kanidmd/src/lib/value.rs
+++ b/kanidmd/src/lib/value.rs
@@ -98,7 +98,7 @@ impl fmt::Display for IndexType {
 }
 
 #[allow(non_camel_case_types)]
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Deserialize, Serialize)]
+#[derive(Hash, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Deserialize, Serialize)]
 pub enum SyntaxType {
     // We need an insensitive string type too ...
     // We also need to "self host" a syntax type, and index type
@@ -240,7 +240,7 @@ impl std::fmt::Debug for DataValue {
     }
 }
 
-#[derive(Debug, Clone, Eq, Ord, PartialOrd, PartialEq, Deserialize, Serialize)]
+#[derive(Hash, Debug, Clone, Eq, Ord, PartialOrd, PartialEq, Deserialize, Serialize)]
 pub enum PartialValue {
     Utf8(String),
     Iutf8(String),