Id assignment

coco.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+while true; do
+    cargo test store_test -- --nocapture
+    exit_code=$?
+    if [ $exit_code -ne 0 ]; then
+        break
+    fi
+done

src/backend/foundationdb/bitmap.rs

@@ -1,15 +1,17 @@
+use ahash::AHashSet;
 use roaring::RoaringBitmap;
 
-const BITS: u32 = 128;
+const WORD_SIZE_BITS: u32 = 128;
-const WORD_SIZE: u32 = 8;
+const WORD_SIZE: usize = std::mem::size_of::<u128>();
-pub const BITS_PER_BLOCK: u32 = BITS * WORD_SIZE;
+const WORDS_PER_BLOCK: u32 = 8;
+pub const BITS_PER_BLOCK: u32 = WORD_SIZE_BITS * WORDS_PER_BLOCK;
 const BITS_MASK: u32 = BITS_PER_BLOCK - 1;
 
 pub struct DenseBitmap {
     restore_value: u8,
     restore_pos: usize,
     pub block_num: u32,
-    pub bitmap: [u8; std::mem::size_of::<u128>() * WORD_SIZE as usize],
+    pub bitmap: [u8; WORD_SIZE * WORDS_PER_BLOCK as usize],
 }
 
 impl DenseBitmap {
@@ -18,7 +20,7 @@ impl DenseBitmap {
             block_num: 0,
             restore_pos: 0,
             restore_value: 0,
-            bitmap: [0; std::mem::size_of::<u128>() * WORD_SIZE as usize],
+            bitmap: [0; WORD_SIZE * WORDS_PER_BLOCK as usize],
         }
     }
 
@@ -27,7 +29,7 @@ impl DenseBitmap {
             block_num: 0,
             restore_pos: 0,
             restore_value: u8::MAX,
-            bitmap: [u8::MAX; std::mem::size_of::<u128>() * WORD_SIZE as usize],
+            bitmap: [u8::MAX; WORD_SIZE * WORDS_PER_BLOCK as usize],
         }
     }
 
@@ -61,37 +63,75 @@ impl DenseBitmap {
 
 pub trait DeserializeBlock {
     fn deserialize_block(&mut self, bytes: &[u8], block_num: u32);
+    fn deserialize_word(&mut self, word: &[u8], block_num: u32, word_num: u32);
+}
+
+pub fn next_available_index(
+    bytes: &[u8],
+    block_num: u32,
+    reserved_ids: &AHashSet<u32>,
+) -> Option<u32> {
+    'outer: for (byte_pos, byte) in bytes.iter().enumerate() {
+        if *byte != u8::MAX {
+            let mut index = 0;
+            loop {
+                while (byte >> index) & 1 == 1 {
+                    index += 1;
+                    if index == 8 {
+                        continue 'outer;
+                    }
+                }
+
+                let id = (block_num * BITS_PER_BLOCK) + ((byte_pos * 8) + index) as u32;
+                if !reserved_ids.contains(&id) {
+                    return Some(id);
+                } else if index < 7 {
+                    index += 1;
+                    continue;
+                } else {
+                    continue 'outer;
+                }
+            }
+        }
+    }
+
+    None
 }
 
 impl DeserializeBlock for RoaringBitmap {
     fn deserialize_block(&mut self, bytes: &[u8], block_num: u32) {
-        debug_assert_eq!(
+        debug_assert_eq!(bytes.len(), WORD_SIZE * WORDS_PER_BLOCK as usize);
-            bytes.len(),
-            std::mem::size_of::<u128>() * WORD_SIZE as usize
-        );
 
-        for (word_num, word) in bytes.chunks_exact(std::mem::size_of::<u128>()).enumerate() {
+        self.deserialize_word(&bytes[..WORD_SIZE], block_num, 0);
-            match u128::from_le_bytes(word.try_into().unwrap()) {
+        self.deserialize_word(&bytes[WORD_SIZE..WORD_SIZE * 2], block_num, 1);
-                0 => continue,
+        self.deserialize_word(&bytes[WORD_SIZE * 2..WORD_SIZE * 3], block_num, 2);
-                u128::MAX => {
+        self.deserialize_word(&bytes[WORD_SIZE * 3..WORD_SIZE * 4], block_num, 3);
-                    self.insert_range(
+        self.deserialize_word(&bytes[WORD_SIZE * 4..WORD_SIZE * 5], block_num, 4);
-                        block_num * BITS_PER_BLOCK + word_num as u32 * 128
+        self.deserialize_word(&bytes[WORD_SIZE * 5..WORD_SIZE * 6], block_num, 5);
-                            ..(block_num * BITS_PER_BLOCK + word_num as u32 * 128) + 128,
+        self.deserialize_word(&bytes[WORD_SIZE * 6..WORD_SIZE * 7], block_num, 6);
+        self.deserialize_word(&bytes[WORD_SIZE * 7..], block_num, 7);
+    }
+
+    #[inline(always)]
+    fn deserialize_word(&mut self, word: &[u8], block_num: u32, word_num: u32) {
+        match u128::from_le_bytes(word.try_into().unwrap()) {
+            0 => (),
+            u128::MAX => {
+                self.insert_range(
+                    block_num * BITS_PER_BLOCK + word_num * WORD_SIZE_BITS
+                        ..(block_num * BITS_PER_BLOCK + word_num * WORD_SIZE_BITS) + WORD_SIZE_BITS,
+                );
+            }
+            mut word => {
+                while word != 0 {
+                    let trailing_zeros = word.trailing_zeros();
+                    self.insert(
+                        block_num * BITS_PER_BLOCK + word_num * WORD_SIZE_BITS + trailing_zeros,
                     );
-                }
+                    word ^= 1 << trailing_zeros;
-                mut word => {
-                    while word != 0 {
-                        let trailing_zeros = word.trailing_zeros();
-                        self.insert(
-                            block_num * BITS_PER_BLOCK + word_num as u32 * 128 + trailing_zeros,
-                        );
-                        word ^= 1 << trailing_zeros;
-                    }
                 }
             }
         }
-
-        //println!("deserializing block {} {}", block_num, self.len());
     }
 }
 
@@ -99,9 +139,12 @@ impl DeserializeBlock for RoaringBitmap {
 mod tests {
     use std::collections::HashMap;
 
+    use ahash::AHashSet;
     use roaring::RoaringBitmap;
 
-    use crate::backend::foundationdb::bitmap::{DenseBitmap, DeserializeBlock, BITS_PER_BLOCK};
+    use crate::backend::foundationdb::bitmap::{
+        next_available_index, DenseBitmap, DeserializeBlock, BITS_PER_BLOCK,
+    };
 
     #[test]
     fn serialize_bitmap_block() {
@@ -123,4 +166,25 @@ mod tests {
             assert_eq!(bitmap, bitmap_blocks);
         }
     }
+
+    #[test]
+    fn get_next_available_index() {
+        let eh = AHashSet::new();
+        let mut uh = AHashSet::new();
+        let mut bm = DenseBitmap::empty();
+        for id in 0..1024 {
+            uh.insert(id);
+            assert_eq!(
+                next_available_index(&bm.bitmap, 0, &eh),
+                Some(id),
+                "failed for {id}"
+            );
+            assert_eq!(
+                next_available_index(&bm.bitmap, 0, &uh),
+                if id < 1023 { Some(id + 1) } else { None },
+                "reserved id failed for {id}"
+            );
+            bm.set_or(id);
+        }
+    }
 }

src/backend/foundationdb/read.rs

@@ -1,5 +1,5 @@
 use std::{
-    ops::{BitAndAssign, BitOrAssign},
+    ops::BitAndAssign,
     time::{Duration, Instant},
 };
 
@@ -74,24 +74,26 @@ impl ReadTransaction<'_> {
         mut key: BitmapKey<T>,
         bm: &mut RoaringBitmap,
     ) -> crate::Result<()> {
-        let from_key = key.serialize();
+        let begin = key.serialize();
         key.block_num = u32::MAX;
-        let to_key = key.serialize();
+        let end = key.serialize();
-        let opt = RangeOption {
+        let mut values = self.trx.get_ranges(
-            mode: StreamingMode::WantAll,
+            RangeOption {
-            reverse: false,
+                begin: KeySelector::first_greater_or_equal(begin),
-            ..RangeOption::from((from_key.as_ref(), to_key.as_ref()))
+                end: KeySelector::first_greater_or_equal(end),
-        };
+                mode: StreamingMode::WantAll,
-        let mut values = self.trx.get_ranges(opt, true);
+                reverse: false,
+                ..RangeOption::default()
+            },
+            true,
+        );
 
         while let Some(values) = values.next().await {
             for value in values? {
                 let key = value.key();
                 bm.deserialize_block(
                     value.value(),
-                    value
+                    key.deserialize_be_u32(key.len() - std::mem::size_of::<u32>())?,
-                        .key()
-                        .deserialize_be_u32(key.len() - std::mem::size_of::<u32>())?,
                 );
             }
         }

src/backend/foundationdb/write.rs

@@ -1,13 +1,34 @@
-use std::time::Instant;
+use std::time::{Duration, Instant, SystemTime};
 
-use foundationdb::{options::MutationType, FdbError};
+use ahash::AHashSet;
+use foundationdb::{
+    options::{MutationType, StreamingMode},
+    FdbError, KeySelector, RangeOption,
+};
+use futures::StreamExt;
+use rand::Rng;
 
 use crate::{
-    write::{Batch, Operation},
+    write::{
-    AclKey, BitmapKey, BlobKey, IndexKey, LogKey, Serialize, Store, ValueKey,
+        key::{DeserializeBigEndian, KeySerializer},
+        Batch, Operation,
+    },
+    AclKey, BitmapKey, BlobKey, Deserialize, IndexKey, LogKey, Serialize, Store, ValueKey,
+    BM_DOCUMENT_IDS,
 };
 
-use super::bitmap::DenseBitmap;
+use super::{
+    bitmap::{next_available_index, DenseBitmap, BITS_PER_BLOCK},
+    SUBSPACE_VALUES,
+};
+
+#[cfg(test)]
+const ID_ASSIGNMENT_EXPIRY: u64 = 2; // seconds
+#[cfg(not(test))]
+pub const ID_ASSIGNMENT_EXPIRY: u64 = 60 * 60; // seconds
+
+const MAX_COMMIT_ATTEMPTS: u8 = 10;
+const MAX_COMMIT_TIME: Duration = Duration::from_secs(10);
 
 impl Store {
     pub async fn write(&self, batch: Batch) -> crate::Result<()> {
@@ -147,20 +168,212 @@ impl Store {
 
             match trx.commit().await {
                 Ok(_) => {
-                    //println!("Success with id {} block {block_num}", document_id);
                     return Ok(());
                 }
                 Err(err) => {
-                    if retry_count < 10 && start.elapsed().as_secs() < 5 {
+                    if retry_count < MAX_COMMIT_ATTEMPTS && start.elapsed() < MAX_COMMIT_TIME {
-                        println!("Retrying with id {}", document_id);
                         err.on_error().await?;
                         retry_count += 1;
                     } else {
-                        println!("Error with id {}", document_id);
                         return Err(FdbError::from(err).into());
                     }
                 }
             }
         }
     }
+
+    pub async fn assign_document_id(&self, account_id: u32, collection: u8) -> crate::Result<u32> {
+        let start = Instant::now();
+
+        loop {
+            //let mut assign_source = 0;
+            // First try to reuse an expired assigned id
+            let begin = IndexKey {
+                account_id,
+                collection,
+                document_id: 0,
+                field: u8::MAX,
+                key: &[],
+            }
+            .serialize();
+            let end = IndexKey {
+                account_id,
+                collection,
+                document_id: u32::MAX,
+                field: u8::MAX,
+                key: &[],
+            }
+            .serialize();
+            let trx = self.db.create_trx()?;
+
+            let mut values = trx.get_ranges(
+                RangeOption {
+                    begin: KeySelector::first_greater_or_equal(begin),
+                    end: KeySelector::first_greater_or_equal(end),
+                    mode: StreamingMode::Iterator,
+                    reverse: false,
+                    ..RangeOption::default()
+                },
+                true,
+            );
+
+            let expired_timestamp = now() - ID_ASSIGNMENT_EXPIRY;
+            let mut reserved_ids = AHashSet::new();
+            let mut expired_ids = Vec::new();
+            while let Some(values) = values.next().await {
+                for value in values? {
+                    let key = value.key();
+                    let document_id =
+                        key.deserialize_be_u32(key.len() - std::mem::size_of::<u32>())?;
+                    if u64::deserialize(value.value())? <= expired_timestamp {
+                        // Found an expired id, reuse it
+                        expired_ids.push(document_id);
+                    } else {
+                        // Keep track of all reserved ids
+                        reserved_ids.insert(document_id);
+                    }
+                }
+            }
+            drop(values);
+
+            let mut document_id = u32::MAX;
+
+            if !expired_ids.is_empty() {
+                // Obtain a random id from the expired ids
+                if expired_ids.len() > 1 {
+                    document_id = expired_ids[rand::thread_rng().gen_range(0..expired_ids.len())];
+                    //assign_source = 1;
+                } else {
+                    document_id = expired_ids[0];
+                    //assign_source = 2;
+                }
+            } else {
+                // Find the next available id
+                let mut key = BitmapKey {
+                    account_id,
+                    collection,
+                    family: BM_DOCUMENT_IDS,
+                    field: u8::MAX,
+                    key: b"",
+                    block_num: 0,
+                };
+                let begin = key.serialize();
+                key.block_num = u32::MAX;
+                let end = key.serialize();
+                let mut values = trx.get_ranges(
+                    RangeOption {
+                        begin: KeySelector::first_greater_or_equal(begin),
+                        end: KeySelector::first_greater_or_equal(end),
+                        mode: StreamingMode::Iterator,
+                        reverse: false,
+                        ..RangeOption::default()
+                    },
+                    true,
+                );
+
+                'outer: while let Some(values) = values.next().await {
+                    for value in values? {
+                        let key = value.key();
+                        if let Some(next_id) = next_available_index(
+                            value.value(),
+                            key.deserialize_be_u32(key.len() - std::mem::size_of::<u32>())?,
+                            &reserved_ids,
+                        ) {
+                            document_id = next_id;
+                            //assign_source = 3;
+
+                            break 'outer;
+                        }
+                    }
+                }
+            }
+
+            // If no ids were found, assign the first available id that is not reserved
+            if document_id == u32::MAX {
+                document_id = 1024;
+                for document_id_ in 0..BITS_PER_BLOCK {
+                    if !reserved_ids.contains(&document_id_) {
+                        document_id = document_id_;
+                        //assign_source = 4;
+
+                        break;
+                    }
+                }
+            }
+
+            // Reserve the id
+            let key = IndexKey {
+                account_id,
+                collection,
+                document_id,
+                field: u8::MAX,
+                key: &[],
+            }
+            .serialize();
+            trx.get(&key, false).await?;
+            trx.set(&key, &now().serialize());
+
+            match trx.commit().await {
+                Ok(_) => {
+                    //println!("assigned id: {document_id} {assign_source}");
+
+                    return Ok(document_id);
+                }
+                Err(err) => {
+                    if start.elapsed() < MAX_COMMIT_TIME {
+                        err.on_error().await?;
+                    } else {
+                        return Err(FdbError::from(err).into());
+                    }
+                }
+            }
+        }
+    }
+
+    pub async fn assign_change_id(&self, account_id: u32, collection: u8) -> crate::Result<u64> {
+        let start = Instant::now();
+        let counter = KeySerializer::new(std::mem::size_of::<u32>() + 2)
+            .write(SUBSPACE_VALUES)
+            .write_leb128(account_id)
+            .write(collection)
+            .finalize();
+
+        loop {
+            // Read id
+            let trx = self.db.create_trx()?;
+            let id = if let Some(bytes) = trx.get(&counter, false).await? {
+                u64::deserialize(&bytes)? + 1
+            } else {
+                0
+            };
+            trx.set(&counter, &id.serialize());
+
+            match trx.commit().await {
+                Ok(_) => {
+                    return Ok(id);
+                }
+                Err(err) => {
+                    if start.elapsed() < MAX_COMMIT_TIME {
+                        err.on_error().await?;
+                    } else {
+                        return Err(FdbError::from(err).into());
+                    }
+                }
+            }
+        }
+    }
+
+    #[cfg(test)]
+    pub async fn destroy(&self) {
+        let trx = self.db.create_trx().unwrap();
+        trx.clear_range(&[0u8], &[u8::MAX]);
+        trx.commit().await.unwrap();
+    }
+}
+
+#[inline(always)]
+fn now() -> u64 {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map_or(0, |d| d.as_secs())
 }

src/fts/bloom.rs

@@ -7,7 +7,7 @@ use std::{
 use roaring::RoaringBitmap;
 use utils::codec::leb128::{Leb128Reader, Leb128Vec};
 
-use crate::{BitmapKey, Deserialize, Error, Serialize, BLOOM_UNIGRAM, BM_BLOOM};
+use crate::{Deserialize, Error, Serialize};
 
 use super::{stemmer::StemmedToken, tokenizers::Token};
 
@@ -102,11 +102,6 @@ impl BloomHash {
     pub fn hash<T: Hash + AsRef<[u8]> + ?Sized>(item: &T) -> Self {
         let h1 = xxhash_rust::xxh3::xxh3_64(item.as_ref());
         let h2 = farmhash::hash64(item.as_ref());
-        /*let h2 = naive_cityhash::cityhash64_with_seeds(
-            item.as_ref(),
-            0x99693e7c5b56f555,
-            0x34809fd70b6ebf45,
-        );*/
         let h3 = AHASHER.hash_one(item);
         let mut sh = *SIPHASHER;
         sh.write(item.as_ref());
@@ -116,39 +111,50 @@ impl BloomHash {
             h: [h1, h2, h3, h4, h1 ^ h2, h2 ^ h3, h3 ^ h4],
         }
     }
-
-    pub fn to_bitmap_key(&self, account_id: u32, collection: u8, field: u8) -> BitmapKey<Vec<u8>> {
-        let mut key = Vec::with_capacity(12);
-        key.extend_from_slice(&self.h[0].to_le_bytes()[..3]);
-        key.extend_from_slice(&self.h[1].to_le_bytes()[..3]);
-        key.extend_from_slice(&self.h[2].to_le_bytes()[..3]);
-        key.extend_from_slice(&self.h[3].to_le_bytes()[..3]);
-
-        BitmapKey {
-            account_id,
-            collection,
-            family: BM_BLOOM | BLOOM_UNIGRAM,
-            field,
-            block_num: 0,
-            key,
-        }
-    }
 }
 
 pub fn hash_token(item: &str) -> Vec<u8> {
-    let h1 = xxhash_rust::xxh3::xxh3_64(item.as_ref());
+    let h1 = xxhash_rust::xxh3::xxh3_64(item.as_ref()).to_le_bytes();
-    let h2 = farmhash::hash64(item.as_ref());
+    let h2 = farmhash::hash64(item.as_ref()).to_le_bytes();
-    let h3 = AHASHER.hash_one(item);
+    let h3 = AHASHER.hash_one(item).to_le_bytes();
     let mut sh = *SIPHASHER;
     sh.write(item.as_ref());
-    let h4 = sh.finish();
+    let h4 = sh.finish().to_le_bytes();
 
-    let mut hash = Vec::with_capacity(12);
+    match item.len() {
-    hash.extend_from_slice(&h1.to_le_bytes()[..3]);
+        0..=8 => {
-    hash.extend_from_slice(&h2.to_le_bytes()[..3]);
+            let mut hash = Vec::with_capacity(6);
-    hash.extend_from_slice(&h3.to_le_bytes()[..3]);
+            hash.extend_from_slice(&h1[..2]);
-    hash.extend_from_slice(&h4.to_le_bytes()[..3]);
+            hash.extend_from_slice(&h2[..2]);
-    hash
+            hash.push(h3[0]);
+            hash.push(h4[0]);
+            hash
+        }
+        9..=16 => {
+            let mut hash = Vec::with_capacity(8);
+            hash.extend_from_slice(&h1[..2]);
+            hash.extend_from_slice(&h2[..2]);
+            hash.extend_from_slice(&h3[..2]);
+            hash.extend_from_slice(&h4[..2]);
+            hash
+        }
+        17..=32 => {
+            let mut hash = Vec::with_capacity(12);
+            hash.extend_from_slice(&h1[..3]);
+            hash.extend_from_slice(&h2[..3]);
+            hash.extend_from_slice(&h3[..3]);
+            hash.extend_from_slice(&h4[..3]);
+            hash
+        }
+        _ => {
+            let mut hash = Vec::with_capacity(16);
+            hash.extend_from_slice(&h1[..4]);
+            hash.extend_from_slice(&h2[..4]);
+            hash.extend_from_slice(&h3[..4]);
+            hash.extend_from_slice(&h4[..4]);
+            hash
+        }
+    }
 }
 
 impl From<&str> for BloomHash {

src/fts/builder.rs

@@ -4,7 +4,7 @@ use ahash::AHashSet;
 
 use crate::{
     write::{BatchBuilder, IntoOperations, Operation},
-    Serialize, BLOOM_BIGRAM, BLOOM_TRIGRAM, BLOOM_UNIGRAM, BLOOM_UNIGRAM_STEM, BM_BLOOM,
+    Serialize, BLOOM_BIGRAM, BLOOM_TRIGRAM, BM_HASH, HASH_EXACT, HASH_STEMMED,
 };
 
 use super::{
@@ -15,7 +15,8 @@ use super::{
     Language,
 };
 
-pub const MAX_TOKEN_LENGTH: usize = 50;
+pub const MAX_TOKEN_LENGTH: usize = (u8::MAX >> 2) as usize;
+pub const MAX_TOKEN_MASK: usize = MAX_TOKEN_LENGTH - 1;
 
 struct Text<'x> {
     field: u8,
@@ -73,41 +74,32 @@ impl<'x> IntoOperations for FtsIndexBuilder<'x> {
             let mut phrase_words = Vec::new();
 
             for token in Stemmer::new(&part.text, language, MAX_TOKEN_LENGTH).collect::<Vec<_>>() {
-                unique_words.insert((token.word.to_string(), BM_BLOOM | BLOOM_UNIGRAM));
+                unique_words.insert((token.word.to_string(), HASH_EXACT));
                 if let Some(stemmed_word) = token.stemmed_word {
-                    unique_words.insert((stemmed_word.into_owned(), BM_BLOOM | BLOOM_UNIGRAM_STEM));
+                    unique_words.insert((stemmed_word.into_owned(), HASH_STEMMED));
                 }
                 phrase_words.push(token.word);
             }
 
-            //let mut bloom_unigram = BloomFilter::new(unique_words.len());
             for (word, family) in unique_words {
-                //let hash = BloomHash::from(word);
-                //bloom_unigram.insert(&hash);
                 batch.ops.push(Operation::Bitmap {
-                    family,
+                    family: BM_HASH | family | (word.len() & MAX_TOKEN_MASK) as u8,
                     field: part.field,
                     key: hash_token(&word),
                     set: true,
                 });
             }
 
-            /*batch.ops.push(Operation::Value {
-                field: part.field,
-                family: BM_BLOOM | BLOOM_UNIGRAM,
-                set: bloom_unigram.serialize().into(),
-            });*/
-
             if phrase_words.len() > 1 {
                 batch.ops.push(Operation::Value {
                     field: part.field,
-                    family: BM_BLOOM | BLOOM_BIGRAM,
+                    family: BLOOM_BIGRAM,
                     set: BloomFilter::to_ngrams(&phrase_words, 2).serialize().into(),
                 });
                 if phrase_words.len() > 2 {
                     batch.ops.push(Operation::Value {
                         field: part.field,
-                        family: BM_BLOOM | BLOOM_TRIGRAM,
+                        family: BLOOM_TRIGRAM,
                         set: BloomFilter::to_ngrams(&phrase_words, 3).serialize().into(),
                     });
                 }
@@ -117,45 +109,3 @@ impl<'x> IntoOperations for FtsIndexBuilder<'x> {
         Ok(())
     }
 }
-
-/*
-impl IntoOperations for TokenIndex {
-    fn build(self, batch: &mut BatchBuilder) -> crate::Result<()> {
-        let mut tokens = AHashSet::new();
-
-        for term in self.terms {
-            for (term_ids, is_exact) in [(term.exact_terms, true), (term.stemmed_terms, false)] {
-                for term_id in term_ids {
-                    tokens.insert((
-                        term.field_id,
-                        is_exact,
-                        self.tokens
-                            .get(term_id as usize)
-                            .ok_or_else(|| {
-                                Error::InternalError("Corrupted term index.".to_string())
-                            })?
-                            .as_bytes()
-                            .to_vec(),
-                    ));
-                }
-            }
-        }
-
-        for (field, is_exact, key) in tokens {
-            batch.ops.push(Operation::Bitmap {
-                family: BM_TERM | if is_exact { TERM_EXACT } else { TERM_STEMMED },
-                field,
-                key,
-                set: false,
-            });
-        }
-
-        batch.ops.push(Operation::Value {
-            field: u8::MAX,
-            set: None,
-        });
-
-        Ok(())
-    }
-}
-*/

src/fts/mod.rs

@@ -21,6 +21,10 @@
  * for more details.
 */
 
+use crate::{BitmapKey, BM_HASH};
+
+use self::{bloom::hash_token, builder::MAX_TOKEN_MASK};
+
 pub mod lang;
 //pub mod pdf;
 pub mod bloom;
@@ -32,8 +36,6 @@ pub mod stemmer;
 //pub mod term_index;
 pub mod tokenizers;
 
-pub const HIGH_RANK_MOD: u64 = 10_240;
-
 #[derive(Debug, PartialEq, Clone, Copy, Hash, Eq, serde::Serialize, serde::Deserialize)]
 pub enum Language {
     Esperanto = 0,
@@ -164,3 +166,16 @@ impl Language {
         .into()
     }
 }
+
+impl BitmapKey<Vec<u8>> {
+    pub fn hash(word: &str, account_id: u32, collection: u8, family: u8, field: u8) -> Self {
+        BitmapKey {
+            account_id,
+            collection,
+            family: BM_HASH | family | (word.len() & MAX_TOKEN_MASK) as u8,
+            field,
+            block_num: 0,
+            key: hash_token(word),
+        }
+    }
+}

src/fts/query.rs

@@ -3,22 +3,22 @@ use std::time::Instant;
 use roaring::RoaringBitmap;
 
 use crate::{
+    backend::foundationdb::read::ReadTransaction,
     fts::{
-        bloom::{hash_token, BloomFilter, BloomHash, BloomHashGroup},
+        bloom::{BloomFilter, BloomHashGroup},
         builder::MAX_TOKEN_LENGTH,
         ngram::ToNgrams,
         stemmer::Stemmer,
         tokenizers::Tokenizer,
     },
-    BitmapKey, Store, ValueKey, BLOOM_BIGRAM, BLOOM_TRIGRAM, BLOOM_UNIGRAM, BLOOM_UNIGRAM_STEM,
+    BitmapKey, ValueKey, BLOOM_BIGRAM, BLOOM_TRIGRAM, HASH_EXACT, HASH_STEMMED,
-    BM_BLOOM,
 };
 
 use super::Language;
 
-impl Store {
+impl ReadTransaction<'_> {
     pub(crate) async fn fts_query(
-        &self,
+        &mut self,
         account_id: u32,
         collection: u8,
         field: u8,
@@ -27,21 +27,25 @@ impl Store {
         match_phrase: bool,
     ) -> crate::Result<Option<RoaringBitmap>> {
         let real_now = Instant::now();
-        let mut trx = self.read_transaction().await?;
 
         let (bitmaps, hashes, family) = if match_phrase {
             let mut tokens = Vec::new();
             let mut bit_keys = Vec::new();
             for token in Tokenizer::new(text, language, MAX_TOKEN_LENGTH) {
-                let hash = BloomHash::from(token.word.as_ref());
+                let key = BitmapKey::hash(
-                let key = hash.to_bitmap_key(account_id, collection, field);
+                    token.word.as_ref(),
+                    account_id,
+                    collection,
+                    HASH_EXACT,
+                    field,
+                );
                 if !bit_keys.contains(&key) {
                     bit_keys.push(key);
                 }
 
                 tokens.push(token.word);
             }
-            let bitmaps = match trx.get_bitmaps_intersection(bit_keys).await? {
+            let bitmaps = match self.get_bitmaps_intersection(bit_keys).await? {
                 Some(b) if !b.is_empty() => b,
                 _ => return Ok(None),
             };
@@ -52,48 +56,32 @@ impl Store {
                 2 => (
                     bitmaps,
                     <Vec<BloomHashGroup>>::to_ngrams(&tokens, 2),
-                    BM_BLOOM | BLOOM_BIGRAM,
+                    BLOOM_BIGRAM,
                 ),
                 _ => (
                     bitmaps,
                     <Vec<BloomHashGroup>>::to_ngrams(&tokens, 3),
-                    BM_BLOOM | BLOOM_TRIGRAM,
+                    BLOOM_TRIGRAM,
                 ),
             }
         } else {
             let mut bitmaps = RoaringBitmap::new();
 
             for token in Stemmer::new(text, language, MAX_TOKEN_LENGTH) {
-                let token1 = hash_token(&token.word);
+                let token1 =
+                    BitmapKey::hash(&token.word, account_id, collection, HASH_EXACT, field);
                 let token2 = if let Some(stemmed_word) = token.stemmed_word {
-                    hash_token(&stemmed_word)
+                    BitmapKey::hash(&stemmed_word, account_id, collection, HASH_STEMMED, field)
                 } else {
-                    token1.clone()
+                    let mut token2 = token1.clone();
+                    token2.family &= !HASH_EXACT;
+                    token2.family |= HASH_STEMMED;
+                    token2
                 };
 
-                trx.refresh_if_old().await?;
+                self.refresh_if_old().await?;
 
-                match trx
+                match self.get_bitmaps_union(vec![token1, token2]).await? {
-                    .get_bitmaps_union(vec![
-                        BitmapKey {
-                            account_id,
-                            collection,
-                            family: BM_BLOOM | BLOOM_UNIGRAM,
-                            field,
-                            block_num: 0,
-                            key: token1,
-                        },
-                        BitmapKey {
-                            account_id,
-                            collection,
-                            family: BM_BLOOM | BLOOM_UNIGRAM_STEM,
-                            field,
-                            block_num: 0,
-                            key: token2,
-                        },
-                    ])
-                    .await?
-                {
                     Some(b) if !b.is_empty() => {
                         if !bitmaps.is_empty() {
                             bitmaps &= b;
@@ -115,9 +103,9 @@ impl Store {
 
         let mut bm = RoaringBitmap::new();
         for document_id in bitmaps {
-            trx.refresh_if_old().await?;
+            self.refresh_if_old().await?;
 
-            if let Some(bloom) = trx
+            if let Some(bloom) = self
                 .get_value::<BloomFilter>(ValueKey {
                     account_id,
                     collection,

src/lib.rs

@@ -94,20 +94,16 @@ pub enum Error {
 }
 
 pub const BM_DOCUMENT_IDS: u8 = 0;
-pub const BM_TERM: u8 = 0x10;
+pub const BM_KEYWORD: u8 = 1 << 5;
-pub const BM_TAG: u8 = 0x20;
+pub const BM_TAG: u8 = 1 << 6;
-pub const BM_BLOOM: u8 = 0x40;
+pub const BM_HASH: u8 = 1 << 7;
 
-pub const BLOOM_UNIGRAM: u8 = 0x00;
+pub const HASH_EXACT: u8 = 0;
-pub const BLOOM_UNIGRAM_STEM: u8 = 0x01;
+pub const HASH_STEMMED: u8 = 1 << 6;
-pub const BLOOM_BIGRAM: u8 = 0x02;
-pub const BLOOM_TRIGRAM: u8 = 0x04;
 
-pub const TERM_EXACT: u8 = 0x00;
+pub const BLOOM_BIGRAM: u8 = 1 << 0;
-pub const TERM_STEMMED: u8 = 0x01;
+pub const BLOOM_TRIGRAM: u8 = 1 << 1;
-pub const TERM_STRING: u8 = 0x02;
-pub const TERM_HASH: u8 = 0x04;
 
-pub const TAG_ID: u8 = 0x00;
+pub const TAG_ID: u8 = 0;
-pub const TAG_TEXT: u8 = 0x01;
+pub const TAG_TEXT: u8 = 1 << 0;
-pub const TAG_STATIC: u8 = 0x02;
+pub const TAG_STATIC: u8 = 1 << 1;

src/query/filter.rs

@@ -2,7 +2,7 @@ use std::ops::{BitAndAssign, BitOrAssign, BitXorAssign};
 
 use roaring::RoaringBitmap;
 
-use crate::{write::Tokenize, BitmapKey, Store, BM_TERM, TERM_EXACT};
+use crate::{write::Tokenize, BitmapKey, Store, BM_KEYWORD};
 
 use super::{Filter, ResultSet};
 
@@ -44,7 +44,7 @@ impl Store {
                     trx.get_bitmap(BitmapKey {
                         account_id,
                         collection,
-                        family: BM_TERM | TERM_EXACT,
+                        family: BM_KEYWORD,
                         field,
                         key: value.as_bytes(),
                         #[cfg(feature = "foundation")]
@@ -60,7 +60,7 @@ impl Store {
                             .map(|key| BitmapKey {
                                 account_id,
                                 collection,
-                                family: BM_TERM | TERM_EXACT,
+                                family: BM_KEYWORD,
                                 field,
                                 key: key.into_bytes(),
                                 #[cfg(feature = "foundation")]
@@ -80,7 +80,7 @@ impl Store {
                     language,
                     match_phrase,
                 } => {
-                    self.fts_query(account_id, collection, field, &text, language, match_phrase)
+                    trx.fts_query(account_id, collection, field, &text, language, match_phrase)
                         .await?
                 }
                 Filter::InBitmap { family, field, key } => {

src/tests/assign_id.rs

@@ -0,0 +1,129 @@
+use std::{collections::HashSet, sync::Arc, time::Duration};
+
+use ahash::AHashSet;
+
+use crate::{write::BatchBuilder, Store};
+
+pub async fn test(db: Arc<Store>) {
+    test_1(db.clone()).await;
+    test_2(db.clone()).await;
+    test_3(db.clone()).await;
+    test_4(db).await;
+}
+
+async fn test_1(db: Arc<Store>) {
+    // Test change id assignment
+    let mut handles = Vec::new();
+    let mut expected_ids = HashSet::new();
+
+    // Create 100 change ids concurrently
+    for id in 0..100 {
+        handles.push({
+            let db = db.clone();
+            tokio::spawn(async move { db.assign_change_id(0, 0).await })
+        });
+        expected_ids.insert(id);
+    }
+
+    for handle in handles {
+        let assigned_id = handle.await.unwrap().unwrap();
+        assert!(
+            expected_ids.remove(&assigned_id),
+            "already assigned or invalid: {assigned_id} "
+        );
+    }
+    db.destroy().await;
+}
+
+async fn test_2(db: Arc<Store>) {
+    // Test document id assignment
+    for wait_for_expiry in [true, false] {
+        let mut handles = Vec::new();
+        let mut expected_ids = HashSet::new();
+
+        // Create 100 ids concurrently
+        for id in 0..100 {
+            handles.push({
+                let db = db.clone();
+                tokio::spawn(async move { db.assign_document_id(0, 0).await })
+            });
+            expected_ids.insert(id);
+        }
+
+        for handle in handles {
+            let assigned_id = handle.await.unwrap().unwrap();
+            //println!("assigned id: {assigned_id} ({wait_for_expiry})");
+            assert!(
+                expected_ids.remove(&assigned_id),
+                "already assigned or invalid: {assigned_id} ({wait_for_expiry})"
+            );
+        }
+        assert_eq!(
+            expected_ids.len(),
+            0,
+            "{expected_ids:?} ({wait_for_expiry})"
+        );
+
+        if wait_for_expiry {
+            tokio::time::sleep(Duration::from_secs(3)).await;
+        }
+    }
+    db.destroy().await;
+}
+
+async fn test_3(db: Arc<Store>) {
+    // Create document ids and try reassigning
+    let mut expected_ids = AHashSet::new();
+    let mut batch = BatchBuilder::new();
+    batch.with_account_id(0).with_collection(0);
+    for pos in 0..100 {
+        let id = db.assign_document_id(0, 0).await.unwrap();
+        if pos % 2 == 0 {
+            batch.create_document(id);
+        } else {
+            expected_ids.insert(id);
+        }
+    }
+    db.write(batch.build()).await.unwrap();
+
+    // Wait for ids to expire
+    tokio::time::sleep(Duration::from_secs(3)).await;
+
+    for _ in 0..expected_ids.len() {
+        let id = db.assign_document_id(0, 0).await.unwrap();
+        assert!(
+            expected_ids.remove(&id),
+            "already assigned or invalid: {id}"
+        );
+    }
+    assert_eq!(db.assign_document_id(0, 0).await.unwrap(), 100);
+    assert_eq!(db.assign_document_id(0, 0).await.unwrap(), 101);
+
+    db.destroy().await;
+}
+
+async fn test_4(db: Arc<Store>) {
+    // Try reassigning deleted ids
+    let mut expected_ids = AHashSet::new();
+    let mut batch = BatchBuilder::new();
+    batch.with_account_id(0).with_collection(0);
+    for id in 0..100 {
+        if id % 2 == 0 {
+            batch.create_document(id);
+        } else {
+            expected_ids.insert(id);
+        }
+    }
+    db.write(batch.build()).await.unwrap();
+    for _ in 0..expected_ids.len() {
+        let id = db.assign_document_id(0, 0).await.unwrap();
+        assert!(
+            expected_ids.remove(&id),
+            "already assigned or invalid: {id}"
+        );
+    }
+    assert_eq!(db.assign_document_id(0, 0).await.unwrap(), 100);
+    assert_eq!(db.assign_document_id(0, 0).await.unwrap(), 101);
+
+    db.destroy().await;
+}

src/tests/mod.rs

@@ -1,19 +1,22 @@
+pub mod assign_id;
 pub mod query;
 
-use std::{collections::BTreeSet, f64::consts::LN_2, io::Read, time::Instant};
+use std::{io::Read, sync::Arc};
-
-use bitpacking::{BitPacker, BitPacker4x, BitPacker8x};
-use rand::Rng;
-use roaring::RoaringBitmap;
-
-use crate::fts::{
-    bloom::BloomFilter,
-    stemmer::{StemmedToken, Stemmer},
-    Language,
-};
 
 use super::*;
 
+#[tokio::test]
+pub async fn store_test() {
+    let db = Arc::new(Store::open().await.unwrap());
+    let insert = true;
+    if insert {
+        db.destroy().await;
+    }
+    assign_id::test(db).await;
+
+    //query::test(db, insert).await;
+}
+
 pub fn deflate_artwork_data() -> Vec<u8> {
     let mut csv_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"));
     csv_path.push("src");
@@ -29,6 +32,7 @@ pub fn deflate_artwork_data() -> Vec<u8> {
     result
 }
 
+/*
 #[test]
 fn it_works() {
     for n in [10, 100, 1000, 5000, 10000, 100000] {
@@ -83,3 +87,4 @@ fn it_works() {
         rb1.serialized_size() as f64 / rb2.serialized_size() as f64
     );*/
 }
+*/

src/tests/query.rs

@@ -32,8 +32,8 @@ use crate::{
     fts::{builder::FtsIndexBuilder, Language},
     query::{Comparator, Filter},
     tests::deflate_artwork_data,
-    write::{BatchBuilder, IntoBitmap, F_INDEX, F_TOKENIZE, F_VALUE},
+    write::{BatchBuilder, F_INDEX, F_TOKENIZE, F_VALUE},
-    Store, ValueKey, BM_DOCUMENT_IDS,
+    Store, ValueKey,
 };
 
 pub const FIELDS: [&str; 20] = [
@@ -91,19 +91,6 @@ const FIELDS_OPTIONS: [FieldType; 20] = [
     FieldType::Text,     // "url",
 ];
 
-#[tokio::test]
-pub async fn db_test() {
-    let db = Arc::new(Store::open().await.unwrap());
-    let insert = false;
-    if insert {
-        let trx = db.db.create_trx().unwrap();
-        trx.clear_range(&[0u8], &[u8::MAX]);
-        trx.commit().await.unwrap();
-    }
-
-    test(db, insert).await;
-}
-
 #[allow(clippy::mutex_atomic)]
 pub async fn test(db: Arc<Store>, do_insert: bool) {
     let pool = rayon::ThreadPoolBuilder::new()

src/write/batch.rs

@@ -1,4 +1,4 @@
-use crate::{BM_DOCUMENT_IDS, BM_TERM, TERM_EXACT};
+use crate::{BM_DOCUMENT_IDS, BM_KEYWORD};
 
 use super::{
     Batch, BatchBuilder, HasFlag, IntoBitmap, IntoOperations, Operation, Serialize, Tokenize,
@@ -24,6 +24,15 @@ impl BatchBuilder {
 
     pub fn create_document(&mut self, document_id: u32) -> &mut Self {
         self.ops.push(Operation::DocumentId { document_id });
+
+        // Remove reserved id
+        self.ops.push(Operation::Index {
+            field: u8::MAX,
+            key: vec![],
+            set: false,
+        });
+
+        // Add document id
         self.ops.push(Operation::Bitmap {
             family: BM_DOCUMENT_IDS,
             field: u8::MAX,
@@ -61,7 +70,7 @@ impl BatchBuilder {
         if options.has_flag(F_TOKENIZE) {
             for token in value.tokenize() {
                 self.ops.push(Operation::Bitmap {
-                    family: BM_TERM | TERM_EXACT,
+                    family: BM_KEYWORD,
                     field,
                     key: token.into_bytes(),
                     set: is_set,

src/write/mod.rs

@@ -110,6 +110,14 @@ impl Deserialize for String {
     }
 }
 
+impl Deserialize for u64 {
+    fn deserialize(bytes: &[u8]) -> crate::Result<Self> {
+        Ok(u64::from_be_bytes(bytes.try_into().map_err(|_| {
+            crate::Error::InternalError("Failed to deserialize u64".to_string())
+        })?))
+    }
+}
+
 trait HasFlag {
     fn has_flag(&self, flag: u32) -> bool;
 }