Repository: meilisearch/meilisearch
File: crates/milli/src/update/upgrade/v1_32.rs (Rust)

Code before:
        }

        let mut iter = db.prefix_iter_mut(wtxn, &(fid_to_delete, 0))?;
        while let Some(((fid, _word_count), ())) = iter.next().transpose()? {
            debug_assert_eq!(fid, fid_to_delete);
            // safety: We don't keep any references to the data.
            unsafe { iter.del_current()? };
        }
    }

    Ok(())
}

/// Fetches keys to delete in parallel by using the FST
/// to balance the work between threads.
fn fetch_keys_to_delete_in_parallel<'txn>(
    wtxn: &mut RwTxn<'txn>,
    index: &Index,
    database: Database<StrBEU16Codec, Unit>,
    fids_to_delete: &BTreeSet<u16>,
) -> Result<LinkedList<Vec<Result<Vec<Box<[u8]>>>>>> {
    let fst = index.words_fst(wtxn)?;

    // TODO get this number from the CLI parameters
    let threads_count = rayon::current_num_threads() * 4;
    let keys_by_thread = (fst.len() / threads_count) + (fst.len() % threads_count);

    // We iterate over the FST keys that represents the word dictionary and
    // roughly represents what can be found in the database we are cleaning.
    //
    // The database we are cleaning contains different words from the word
    // dictionary as it contains words from fields that are not indexed too
    // but it is mixed with indexed ones.
    //
    // We then divide equally the entries of the database to clean by
    // selecting ranges of keys that will be processed by each thread. We
    // also make sure not to specify the first and last keys to make sure
    // that if the fields to clean have keys that are higher or lower than
    // the first or last keys in the word dictionary we still find them.

    // Here We make sure to start with an unbounded
    // left bound for the first range
    let mut bounds = vec![None];
    let mut stream = fst.stream();
    let mut count = 0;
    while let Some(key) = stream.next() {
        let is_first = count == 0;
        let is_last = count == fst.len() - 1;

        // In this loop we make sure to account for every bounds
        // to divide the work between threads but not send the bounds

Review comment (bug):
I think this formula doesn't distribute "evenly" work and can lead to thread underutilization. 

It think this one works better:

```rust
let keys_by_thread = fst.len().div_ceil(threads_count);
```

Let say the FST has 1,000 keys and we have 8 cores. The threadpool will be comprise of 32 threads. With the wrong formula we allocate 39 keys per threads, which utilizes only 25 threads (1,000 keys / 39).

With the other formula, we allocate 32 keys per threads, which utilizes 31 threads (1,000 keys / 32).

Granted, as the FST grows, the difference narrows, but it’s still worth mentioning.

Code after:

fn delete_old_fid_word_count_docids(
    wtxn: &mut RwTxn<'_>,
    index: &Index,
    must_stop_processing: &MustStopProcessing,
    fids_to_delete: &BTreeSet<u16>,
) -> Result<(), Error> {
    let db = index.field_id_word_count_docids.remap_data_type::<DecodeIgnore>();
    for &fid_to_delete in fids_to_delete {
        if must_stop_processing.get() {
            return Err(Error::InternalError(InternalError::AbortedIndexation));
        }

        let mut iter = db.prefix_iter_mut(wtxn, &(fid_to_delete, 0))?;
        while let Some(((fid, _word_count), ())) = iter.next().transpose()? {
            debug_assert_eq!(fid, fid_to_delete);
            // safety: We don't keep any references to the data.
            unsafe { iter.del_current()? };
        }
    }

    Ok(())
}

/// Fetches keys to delete in parallel by using the FST
/// to balance the work between threads.
fn fetch_keys_to_delete_in_parallel<'txn>(
    wtxn: &mut RwTxn<'txn>,
    index: &Index,
    database: Database<StrBEU16Codec, Unit>,
    fids_to_delete: &BTreeSet<u16>,
    bounds: &[Option<Box<[u8]>>],
    pool: &ThreadPool,
) -> Result<LinkedList<Vec<Result<Vec<Box<[u8]>>>>>> {
    // We generate enough read transactions for each thread.
    let rtxns = iter::repeat_with(|| index.env.nested_read_txn(wtxn))
        .take(bounds.len().saturating_sub(1))
        .collect::<heed::Result<Vec<_>>>()?;

    let results = pool.install(|| {
        rtxns
            .into_par_iter()
            .zip_eq(bounds.windows(2).collect::<Vec<_>>())
            .map(|(rtxn, win)| {
                let bound = match [win[0].as_deref(), win[1].as_deref()] {
                    [None, None] => (Bound::Unbounded, Bound::Unbounded),
                    [None, Some(end)] => (Bound::Unbounded, Bound::Excluded(end)),
                    [Some(start), None] => (Bound::Included(start), Bound::Unbounded),
                    [Some(start), Some(end)] => (Bound::Included(start), Bound::Excluded(end)),
                };