thomas.goyne_478

Committed 02 Aug 2024 05:19PM UTC coverage: 91.089% (-0.01%) from 91.1%

Build # thomas.goyne_478

Build Type

Pull #7944

Evergreen

Committed by

tgoyne

Commit Message

Only track pending client resets done by the same core version

If the previous attempt at performing a client reset was done with a different
core version then we should retry the client reset as the new version may have
fixed a bug that made the previous attempt fail (or may be a downgrade to a
version before when the bug was introduced). This also simplifies the tracking
as it means that we don't need to be able to read trackers created by different
versions.

This also means that we can freely change the schema of the table, which this
takes advantage of to drop the unused primary key and make the error required,
as we never actually stored null and the code reading it would have crashed if
it encountered a null error.

Pull Request Pull Request #7944: Only track pending client resets done by the same core version

Run Details

102704 of 181534 branches covered (56.58%)

138 of 153 new or added lines in 10 files covered. (90.2%)

85 existing lines in 16 files now uncovered.

216717 of 237917 relevant lines covered (91.09%)

5947762.1 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.63

/src/realm/index_string.cpp

/*************************************************************************
 *
 * Copyright 2016 Realm Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 **************************************************************************/

#include <cstdio>
#include <iomanip>
#include <list>

#ifdef REALM_DEBUG
#include <iostream>
#endif

#include <realm/exceptions.hpp>
#include <realm/index_string.hpp>
#include <realm/table.hpp>
#include <realm/list.hpp>
#include <realm/timestamp.hpp>
#include <realm/column_integer.hpp>
#include <realm/unicode.hpp>
#include <realm/tokenizer.hpp>

using namespace realm;
using namespace realm::util;

namespace {

void get_child(Array& parent, size_t child_ref_ndx, Array& child) noexcept
{
    ref_type child_ref = parent.get_as_ref(child_ref_ndx);
    child.init_from_ref(child_ref);
    child.set_parent(&parent, child_ref_ndx);
}

// This method reconstructs the string inserted in the search index based on a string
// that matches so far and the last key (only works if complete strings are stored in the index)
static StringData reconstruct_string(size_t offset, StringIndex::key_type key, StringData new_string)
{
    if (key == 0)
        return StringData();

    size_t rest_len = 4;
    char* k = reinterpret_cast<char*>(&key);
    if (k[0] == 'X')
        rest_len = 3;
    else if (k[1] == 'X')
        rest_len = 2;
    else if (k[2] == 'X')
        rest_len = 1;
    else if (k[3] == 'X')
        rest_len = 0;

    REALM_ASSERT(offset + rest_len <= new_string.size());

    return StringData(new_string.data(), offset + rest_len);
}

template <IndexMethod method>
int64_t from_list_full_word(InternalFindResult&, const IntegerColumn&);

template <>
inline int64_t from_list_full_word<index_FindFirst>(InternalFindResult&, const IntegerColumn& key_values)
{
    return key_values.get(0);
}

template <>
inline int64_t from_list_full_word<index_Count>(InternalFindResult&, const IntegerColumn& key_values)
{
    return int64_t(key_values.size());
}

template <>
inline int64_t from_list_full_word<index_FindAll_nocopy>(InternalFindResult& result_ref,
                                                         const IntegerColumn& key_values)
{
    result_ref.payload = from_ref(key_values.get_ref());
    result_ref.start_ndx = 0;
    result_ref.end_ndx = key_values.size();
    return size_t(FindRes_column);
}

} // anonymous namespace

DataType ClusterColumn::get_data_type() const
{
    const Table* table = m_cluster_tree->get_owning_table();
    return table->get_column_type(m_column_key);
}

Mixed ClusterColumn::get_value(ObjKey key) const
{
    const Obj obj{m_cluster_tree->get(key)};
    return obj.get_any(m_column_key);
}

Lst<String> ClusterColumn::get_list(ObjKey key) const
{
    const Obj obj{m_cluster_tree->get(key)};
    return obj.get_list<String>(m_column_key);
}

std::vector<ObjKey> ClusterColumn::get_all_keys() const
{
    std::vector<ObjKey> ret;
    ret.reserve(m_cluster_tree->size());
    m_cluster_tree->traverse([&ret](const Cluster* cluster) {
        auto sz = cluster->node_size();
        for (size_t i = 0; i < sz; i++) {
            ret.push_back(cluster->get_real_key(i));
        }

        return IteratorControl::AdvanceToNext;
    });
    return ret;
}

template <>
int64_t IndexArray::from_list<index_FindFirst>(const Mixed& value, InternalFindResult& /* result_ref */,
                                               const IntegerColumn& key_values, const ClusterColumn& column) const
{
    SortedListComparator slc(column);

    IntegerColumn::const_iterator it_end = key_values.cend();
    IntegerColumn::const_iterator lower = slc.find_start_of_unsorted(value, key_values);

    if (lower == it_end)
        return null_key.value;

    int64_t first_key_value = *lower;

    Mixed actual = column.get_value(ObjKey(first_key_value));
    if (actual != value)
        return null_key.value;

    return first_key_value;
}

template <>
int64_t IndexArray::from_list<index_Count>(const Mixed& value, InternalFindResult& /* result_ref */,
                                           const IntegerColumn& key_values, const ClusterColumn& column) const
{
    SortedListComparator slc(column);

    IntegerColumn::const_iterator it_end = key_values.cend();
    IntegerColumn::const_iterator lower = slc.find_start_of_unsorted(value, key_values);
    if (lower == it_end)
        return 0;

    int64_t first_key_value = *lower;

    Mixed actual = column.get_value(ObjKey(first_key_value));
    if (actual != value)
        return 0;

    IntegerColumn::const_iterator upper = slc.find_end_of_unsorted(value, key_values, lower);
    int64_t cnt = upper - lower;

    return cnt;
}

template <>
int64_t IndexArray::from_list<index_FindAll_nocopy>(const Mixed& value, InternalFindResult& result_ref,
                                                    const IntegerColumn& key_values,
                                                    const ClusterColumn& column) const
{
    SortedListComparator slc(column);
    IntegerColumn::const_iterator it_end = key_values.cend();
    IntegerColumn::const_iterator lower = slc.find_start_of_unsorted(value, key_values);

    if (lower == it_end)
        return size_t(FindRes_not_found);

    ObjKey first_key = ObjKey(*lower);

    Mixed actual = column.get_value(ObjKey(first_key));
    if (actual != value)
        return size_t(FindRes_not_found);

    // Optimization: check the last entry before trying upper bound.
    IntegerColumn::const_iterator upper = it_end;
    --upper;
    // Single result if upper matches lower
    if (upper == lower) {
        result_ref.payload = *lower;
        return size_t(FindRes_single);
    }

    // Check string value at upper, if equal return matches in (lower, upper]
    ObjKey last_key = ObjKey(*upper);
    actual = column.get_value(ObjKey(last_key));
    if (actual == value) {
        result_ref.payload = from_ref(key_values.get_ref());
        result_ref.start_ndx = lower.get_position();
        result_ref.end_ndx = upper.get_position() + 1; // one past last match
        return size_t(FindRes_column);
    }

    // Last result is not equal, find the upper bound of the range of results.
    // Note that we are passing upper which is cend() - 1 here as we already
    // checked the last item manually.
    upper = slc.find_end_of_unsorted(value, key_values, lower);

    result_ref.payload = from_ref(key_values.get_ref());
    result_ref.start_ndx = lower.get_position();
    result_ref.end_ndx = upper.get_position();
    return int64_t(FindRes_column);
}


template <IndexMethod method>
int64_t IndexArray::index_string(const Mixed& value, InternalFindResult& result_ref,
                                 const ClusterColumn& column) const
{
    // Return`realm::not_found`, or an index to the (any) match
    constexpr bool first(method == index_FindFirst);
    // Return 0, or the number of items that match the specified `value`
    constexpr bool get_count(method == index_Count);
    // Same as `index_FindAll` but does not copy matching rows into `column`
    // returns FindRes_not_found if there are no matches
    // returns FindRes_single and the row index (literal) in result_ref.payload
    // or returns FindRes_column and the reference to a column of duplicates in
    // result_ref.result with the results in the bounds start_ndx, and end_ndx
    constexpr bool allnocopy(method == index_FindAll_nocopy);

    constexpr int64_t local_not_found = allnocopy ? int64_t(FindRes_not_found) : first ? null_key.value : 0;

    const char* data = m_data;
    const char* header;
    uint_least8_t width = m_width;
    bool is_inner_node = m_is_inner_bptree_node;
    typedef StringIndex::key_type key_type;
    size_t stringoffset = 0;

    StringConversionBuffer buffer;
    StringData index_data = value.get_index_data(buffer);

    // Create 4 byte index key
    key_type key = StringIndex::create_key(index_data, stringoffset);

    for (;;) {
        // Get subnode table
        ref_type offsets_ref = to_ref(get_direct(data, width, 0));

        // Find the position matching the key
        const char* offsets_header = m_alloc.translate(offsets_ref);
        const char* offsets_data = get_data_from_header(offsets_header);
        size_t offsets_size = get_size_from_header(offsets_header);
        size_t pos = ::lower_bound<32>(offsets_data, offsets_size, key); // keys are always 32 bits wide

        // If key is outside range, we know there can be no match
        if (pos == offsets_size)
            return local_not_found;

        // Get entry under key
        size_t pos_refs = pos + 1; // first entry in refs points to offsets
        uint64_t ref = get_direct(data, width, pos_refs);

        if (is_inner_node) {
            // Set vars for next iteration
            header = m_alloc.translate(to_ref(ref));
            data = get_data_from_header(header);
            width = get_width_from_header(header);
            is_inner_node = get_is_inner_bptree_node_from_header(header);
            continue;
        }

        key_type stored_key = key_type(get_direct<32>(offsets_data, pos));

        if (stored_key != key)
            return local_not_found;

        // Literal row index (tagged)
        if (ref & 1) {
            int64_t key_value = int64_t(ref >> 1);

            Mixed a = column.full_word() ? reconstruct_string(stringoffset, key, index_data)
                                         : column.get_value(ObjKey(key_value));
            if (a == value) {
                result_ref.payload = key_value;
                return first ? key_value : get_count ? 1 : FindRes_single;
            }
            return local_not_found;
        }

        const char* sub_header = m_alloc.translate(ref_type(ref));
        const bool sub_isindex = get_context_flag_from_header(sub_header);

        // List of row indices with common prefix up to this point, in sorted order.
        if (!sub_isindex) {
            const IntegerColumn sub(m_alloc, ref_type(ref));
            if (column.full_word()) {
                return from_list_full_word<method>(result_ref, sub);
            }

            return from_list<method>(value, result_ref, sub, column);
        }

        // Recurse into sub-index;
        header = sub_header;
        data = get_data_from_header(header);
        width = get_width_from_header(header);
        is_inner_node = get_is_inner_bptree_node_from_header(header);

        // Go to next key part of the string. If the offset exceeds the string length, the key will be 0
        stringoffset += 4;

        // Update 4 byte index key
        key = StringIndex::create_key(index_data, stringoffset);
    }
}


void IndexArray::from_list_all_ins(StringData upper_value, std::vector<ObjKey>& result, const IntegerColumn& rows,
                                   const ClusterColumn& column) const
{
    // optimization for the most common case, where all the strings under a given subindex are equal
    Mixed first = column.get_value(ObjKey(*rows.cbegin()));
    Mixed last = column.get_value(ObjKey(*(rows.cend() - 1)));
    if (first == last) {
        if (!first.is_type(type_String))
            return;
        auto first_str_upper = case_map(first.get_string(), true);
        if (first_str_upper != upper_value) {
            return;
        }

        size_t sz = result.size() + rows.size();
        result.reserve(sz);
        for (IntegerColumn::const_iterator it = rows.cbegin(); it != rows.cend(); ++it) {
            result.push_back(ObjKey(*it));
        }
        return;
    }

    // special case for very long strings, where they might have a common prefix and end up in the
    // same subindex column, but still not be identical
    for (IntegerColumn::const_iterator it = rows.cbegin(); it != rows.cend(); ++it) {
        ObjKey key = ObjKey(*it);
        Mixed val = column.get_value(key);
        if (val.is_type(type_String)) {
            auto upper_str = case_map(val.get_string(), true);
            if (upper_str == upper_value) {
                result.push_back(key);
            }
        }
    }

    return;
}


void IndexArray::from_list_all(const Mixed& value, std::vector<ObjKey>& result, const IntegerColumn& rows,
                               const ClusterColumn& column) const
{
    if (column.full_word()) {
        result.reserve(rows.size());
        for (IntegerColumn::const_iterator it = rows.cbegin(); it != rows.cend(); ++it) {
            result.push_back(ObjKey(*it));
        }

        return;
    }

    SortedListComparator slc(column);

    IntegerColumn::const_iterator it_end = rows.cend();
    IntegerColumn::const_iterator lower = slc.find_start_of_unsorted(value, rows);
    if (lower == it_end)
        return;

    ObjKey key = ObjKey(*lower);

    Mixed a = column.get_value(key);
    if (a != value)
        return;

    IntegerColumn::const_iterator upper = slc.find_end_of_unsorted(value, rows, lower);

    // Copy all matches into result column
    size_t sz = result.size() + (upper - lower);
    result.reserve(sz);
    for (IntegerColumn::const_iterator it = lower; it != upper; ++it) {
        result.push_back(ObjKey(*it));
    }
}


namespace {

// Helper functions for SearchList (index_string_all_ins) for generating permutations of index keys

// replicates the 4 least significant bits each times 8
// eg: abcd -> aaaaaaaabbbbbbbbccccccccdddddddd
uint32_t replicate_4_lsb_x8(uint32_t i)
{
    REALM_ASSERT_DEBUG(i <= 15);
    i *= 0x204081;
    i &= 0x1010101;
    i *= 0xff;
    return i;
}

int32_t select_from_mask(int32_t a, int32_t b, int32_t mask)
{
    return a ^ ((a ^ b) & mask);
}

// Given upper and lower keys: "ABCD" and "abcd", the 4 LSBs in the permutation argument determine the
// final key:
// Permutation 0  = "ABCD"
// Permutation 1  = "ABCd"
// Permutation 8  = "aBCD"
// Permutation 15 = "abcd"
using key_type = StringIndex::key_type;
static key_type generate_key(key_type upper, key_type lower, int permutation)
{
    return select_from_mask(upper, lower, replicate_4_lsb_x8(permutation));
}


// Helper structure for IndexArray::index_string_all_ins to generate and keep track of search key permutations,
// when traversing the trees.
struct SearchList {
    struct Item {
        const char* header;
        size_t string_offset;
        key_type key;
    };

    SearchList(const util::Optional<std::string>& upper_value, const util::Optional<std::string>& lower_value)
        : m_upper_value(upper_value)
        , m_lower_value(lower_value)
    {
        m_keys_seen.reserve(num_permutations);
    }

    // Add all unique keys for this level to the internal work stack
    void add_all_for_level(const char* header, size_t string_offset)
    {
        m_keys_seen.clear();
        const key_type upper_key = StringIndex::create_key(m_upper_value, string_offset);
        const key_type lower_key = StringIndex::create_key(m_lower_value, string_offset);
        for (int p = 0; p < num_permutations; ++p) {
            // FIXME: This might still be incorrect due to multi-byte unicode characters (crossing the 4 byte key
            // size) being combined incorrectly.
            const key_type key = generate_key(upper_key, lower_key, p);
            const bool new_key = std::find(m_keys_seen.cbegin(), m_keys_seen.cend(), key) == m_keys_seen.cend();
            if (new_key) {
                m_keys_seen.push_back(key);
                add_next(header, string_offset, key);
            }
        }
    }

    bool empty() const
    {
        return m_items.empty();
    }

    Item get_next()
    {
        Item item = m_items.back();
        m_items.pop_back();
        return item;
    }

    // Add a single entry to the internal work stack. Used to traverse the inner trees (same key)
    void add_next(const char* header, size_t string_offset, key_type key)
    {
        m_items.push_back({header, string_offset, key});
    }

private:
    static constexpr int num_permutations = 1 << sizeof(key_type); // 4 bytes gives up to 16 search keys

    std::vector<Item> m_items;

    const util::Optional<std::string> m_upper_value;
    const util::Optional<std::string> m_lower_value;

    std::vector<key_type> m_keys_seen;
};


} // namespace


void IndexArray::index_string_all_ins(StringData value, std::vector<ObjKey>& result,
                                      const ClusterColumn& column) const
{
    REALM_ASSERT(!value.is_null());

    const util::Optional<std::string> upper_value = case_map(value, true);
    const util::Optional<std::string> lower_value = case_map(value, false);
    SearchList search_list(upper_value, lower_value);

    const char* top_header = get_header_from_data(m_data);
    search_list.add_all_for_level(top_header, 0);

    while (!search_list.empty()) {
        SearchList::Item item = search_list.get_next();

        const char* const header = item.header;
        const size_t string_offset = item.string_offset;
        const key_type key = item.key;
        const char* const data = get_data_from_header(header);
        const uint_least8_t width = get_width_from_header(header);
        const bool is_inner_node = get_is_inner_bptree_node_from_header(header);

        // Get subnode table
        ref_type offsets_ref = to_ref(get_direct(data, width, 0));

        // Find the position matching the key
        const char* const offsets_header = m_alloc.translate(offsets_ref);
        const char* const offsets_data = get_data_from_header(offsets_header);
        const size_t offsets_size = get_size_from_header(offsets_header);
        const size_t pos = ::lower_bound<32>(offsets_data, offsets_size, key); // keys are always 32 bits wide

        // If key is outside range, we know there can be no match
        if (pos == offsets_size)
            continue;

        // Get entry under key
        const size_t pos_refs = pos + 1; // first entry in refs points to offsets
        const uint64_t ref = get_direct(data, width, pos_refs);

        if (is_inner_node) {
            // Set vars for next iteration
            const char* const inner_header = m_alloc.translate(to_ref(ref));
            search_list.add_next(inner_header, string_offset, key);
            continue;
        }

        const key_type stored_key = key_type(get_direct<32>(offsets_data, pos));

        if (stored_key != key)
            continue;

        // Literal row index (tagged)
        if (ref & 1) {
            ObjKey k(int64_t(ref >> 1));

            // The buffer is needed when for when this is an integer index.
            StringConversionBuffer buffer;
            const StringData str = column.get_value(k).get_index_data(buffer);
            const util::Optional<std::string> upper_str = case_map(str, true);
            if (upper_str == upper_value) {
                result.push_back(k);
            }
            continue;
        }

        const char* const sub_header = m_alloc.translate(ref_type(ref));
        const bool sub_isindex = get_context_flag_from_header(sub_header);

        // List of row indices with common prefix up to this point, in sorted order.
        if (!sub_isindex) {
            const IntegerColumn sub(m_alloc, ref_type(ref));
            from_list_all_ins(upper_value, result, sub, column);
            continue;
        }

        // Recurse into sub-index;
        search_list.add_all_for_level(sub_header, string_offset + 4);
    }

    // sort the result and return a std::vector
    std::sort(result.begin(), result.end());
}


void IndexArray::index_string_all(const Mixed& value, std::vector<ObjKey>& result, const ClusterColumn& column) const
{
    const char* data = m_data;
    const char* header;
    uint_least8_t width = m_width;
    bool is_inner_node = m_is_inner_bptree_node;
    size_t stringoffset = 0;

    StringConversionBuffer buffer;
    StringData index_data = value.get_index_data(buffer);
    // Create 4 byte index key
    key_type key = StringIndex::create_key(index_data, stringoffset);

    for (;;) {
        // Get subnode table
        ref_type offsets_ref = to_ref(get_direct(data, width, 0));

        // Find the position matching the key
        const char* offsets_header = m_alloc.translate(offsets_ref);
        const char* offsets_data = get_data_from_header(offsets_header);
        size_t offsets_size = get_size_from_header(offsets_header);
        size_t pos = ::lower_bound<32>(offsets_data, offsets_size, key); // keys are always 32 bits wide

        // If key is outside range, we know there can be no match
        if (pos == offsets_size)
            return;

        // Get entry under key
        size_t pos_refs = pos + 1; // first entry in refs points to offsets
        uint64_t ref = get_direct(data, width, pos_refs);

        if (is_inner_node) {
            // Set vars for next iteration
            header = m_alloc.translate(ref_type(ref));
            data = get_data_from_header(header);
            width = get_width_from_header(header);
            is_inner_node = get_is_inner_bptree_node_from_header(header);
            continue;
        }

        key_type stored_key = key_type(get_direct<32>(offsets_data, pos));

        if (stored_key != key)
            return;

        // Literal row index (tagged)
        if (ref & 1) {
            ObjKey k(int64_t(ref >> 1));

            if (column.full_word() || column.get_value(k) == value) {
                result.push_back(k);
                return;
            }
            return;
        }

        const char* sub_header = m_alloc.translate(ref_type(ref));
        const bool sub_isindex = get_context_flag_from_header(sub_header);

        // List of row indices with common prefix up to this point, in sorted order.
        if (!sub_isindex) {
            const IntegerColumn sub(m_alloc, ref_type(ref));
            return from_list_all(value, result, sub, column);
        }

        // Recurse into sub-index;
        header = sub_header;
        data = get_data_from_header(header);
        width = get_width_from_header(header);
        is_inner_node = get_is_inner_bptree_node_from_header(header);

        // Go to next key part of the string. If the offset exceeds the string length, the key will be 0
        stringoffset += 4;

        // Update 4 byte index key
        key = StringIndex::create_key(index_data, stringoffset);
    }
}

static void get_all_keys_below(std::set<int64_t>& result, ref_type ref, Allocator& alloc)
{
    const char* sub_header = alloc.translate(ref_type(ref));
    const bool sub_isindex = NodeHeader::get_context_flag_from_header(sub_header);

    if (sub_isindex) {
        Array tree(alloc);
        tree.init_from_ref(ref);
        auto sz = tree.size();
        for (size_t n = 1; n < sz; n++) {
            auto rot = tree.get_as_ref_or_tagged(n);
            // Literal row index (tagged)
            if (rot.is_tagged()) {
                result.insert(rot.get_as_int());
            }
            else {
                get_all_keys_below(result, rot.get_as_ref(), alloc);
            }
        }
    }
    else {
        IntegerColumn tree(alloc, ref);
        tree.for_all([&result](int64_t i) {
            result.insert(i);
        });
    }
}

void IndexArray::_index_string_find_all_prefix(std::set<int64_t>& result, StringData str, const char* header) const
{
    size_t stringoffset = 0;

    for (;;) {
        const char* data = NodeHeader::get_data_from_header(header);
        uint_least8_t width = get_width_from_header(header);

        // Create 4 byte lower and upper key
        key_type lower = 0;
        size_t i = 0;
        size_t n = str.size() - stringoffset;
        bool is_at_string_end = (n <= 4);
        if (!is_at_string_end) {
            n = 4;
        }
        while (i < n) {
            lower <<= 8;
            lower += str[stringoffset + i++];
        }
        size_t shift = (4 - i) * 8;
        key_type upper = lower + 1;
        lower <<= shift;
        upper <<= shift;
        upper--;

        // Get index array
        ref_type offsets_ref = to_ref(get_direct(data, width, 0));

        // Find the position matching the key
        const char* offsets_header = m_alloc.translate(offsets_ref);
        const char* offsets_data = get_data_from_header(offsets_header);
        size_t offsets_size = get_size_from_header(offsets_header);
        size_t pos = ::lower_bound<32>(offsets_data, offsets_size, lower); // keys are always 32 bits wide
        // If key is outside range, we know there can be no match
        if (pos == offsets_size)
            return;

        size_t pos_refs = pos + 1; // first entry in refs points to offsets

        if (NodeHeader::get_is_inner_bptree_node_from_header(header)) {
            bool done = false;
            while (!done) {
                // Recursively call with child node
                const char* header = m_alloc.translate(to_ref(get_direct(data, width, pos_refs++)));
                _index_string_find_all_prefix(result, str, header);

                // Check if current node is past end of key range or last node
                auto key = get_direct<32>(offsets_data, pos++);
                done = key > upper || pos == offsets_size;
            }
            return;
        }

        size_t end = ::upper_bound<32>(offsets_data, offsets_size, upper); // keys are always 32 bits wide

        if (pos == end) {
            // No match
            return;
        }

        if (is_at_string_end) {
            // now get all entries from start to end
            for (size_t ndx = pos_refs; ndx <= end; ndx++) {
                uint64_t ref = get_direct(data, width, ndx);
                // Literal row index (tagged)
                if (ref & 1) {
                    result.emplace(int64_t(ref >> 1));
                }
                else {
                    get_all_keys_below(result, to_ref(ref), m_alloc);
                }
            }
            return;
        }

        // When we are not at end of string then we are comparing against the whole key
        // and we can have at most one match
        REALM_ASSERT(end == pos + 1);
        header = m_alloc.translate(to_ref(get_direct(data, width, pos_refs)));
        stringoffset += 4;
    }
}

ObjKey IndexArray::index_string_find_first(const Mixed& value, const ClusterColumn& column) const
{
    InternalFindResult unused;
    return ObjKey(index_string<index_FindFirst>(value, unused, column));
}


void IndexArray::index_string_find_all(std::vector<ObjKey>& result, const Mixed& value, const ClusterColumn& column,
                                       bool case_insensitive) const
{
    if (case_insensitive && value.is_type(type_String)) {
        index_string_all_ins(value.get_string(), result, column);
    }
    else {
        index_string_all(value, result, column);
    }
}

FindRes IndexArray::index_string_find_all_no_copy(const Mixed& value, const ClusterColumn& column,
                                                  InternalFindResult& result) const
{
    return static_cast<FindRes>(index_string<index_FindAll_nocopy>(value, result, column));
}

size_t IndexArray::index_string_count(const Mixed& value, const ClusterColumn& column) const
{
    InternalFindResult unused;
    return to_size_t(index_string<index_Count>(value, unused, column));
}

std::unique_ptr<IndexArray> StringIndex::create_node(Allocator& alloc, bool is_leaf)
{
    Array::Type type = is_leaf ? Array::type_HasRefs : Array::type_InnerBptreeNode;
    std::unique_ptr<IndexArray> top(new IndexArray(alloc)); // Throws
    top->create(type);                                      // Throws

    // Mark that this is part of index
    // (as opposed to columns under leaves)
    top->set_context_flag(true);

    // Add subcolumns for leaves
    Array values(alloc);
    values.create(Array::type_Normal);       // Throws
    values.ensure_minimum_width(0x7FFFFFFF); // This ensures 31 bits plus a sign bit
    top->add(values.get_ref());              // first entry in refs points to offsets

    return top;
}

StringIndex::key_type StringIndex::get_last_key() const
{
    Array offsets(m_array->get_alloc());
    get_child(*m_array, 0, offsets);
    return key_type(offsets.back());
}


void StringIndex::insert_with_offset(ObjKey obj_key, StringData index_data, const Mixed& value, size_t offset)
{
    // Create 4 byte index key
    key_type key = create_key(index_data, offset);
    TreeInsert(obj_key, key, offset, index_data, value); // Throws
}

void StringIndex::insert_to_existing_list_at_lower(ObjKey key, Mixed value, IntegerColumn& list,
                                                   const IntegerColumnIterator& lower)
{
    // At this point there exists duplicates of this value, we need to
    // insert value beside it's duplicates so that rows are also sorted
    // in ascending order.
    IntegerColumn::const_iterator upper = [&]() {
        if (m_target_column.full_word()) {
            return list.cend();
        }
        else {
            SortedListComparator slc(m_target_column);
            return slc.find_end_of_unsorted(value, list, lower);
        }
    }();
    // find insert position (the list has to be kept in sorted order)
    // In most cases the refs will be added to the end. So we test for that
    // first to see if we can avoid the binary search for insert position
    IntegerColumn::const_iterator last = upper - ptrdiff_t(1);
    int64_t last_key_value = *last;
    if (key.value >= last_key_value) {
        list.insert(upper.get_position(), key.value);
    }
    else {
        // insert into the group of duplicates, keeping object keys sorted
        IntegerColumn::const_iterator inner_lower = std::lower_bound(lower, upper, key.value);
        if (*inner_lower != key.value) {
            list.insert(inner_lower.get_position(), key.value);
        }
    }
}

void StringIndex::insert_to_existing_list(ObjKey key, Mixed value, IntegerColumn& list)
{
    SortedListComparator slc(m_target_column);
    IntegerColumn::const_iterator it_end = list.cend();
    IntegerColumn::const_iterator lower = slc.find_start_of_unsorted(value, list);

    if (lower == it_end) {
        // Not found and everything is less, just append it to the end.
        list.add(key.value);
    }
    else {
        ObjKey lower_key = ObjKey(*lower);
        Mixed lower_value = get(lower_key);

        if (lower_value != value) {
            list.insert(lower.get_position(), key.value);
        }
        else {
            // At this point there exists duplicates of this value, we need to
            // insert value beside it's duplicates so that rows are also sorted
            // in ascending order.
            insert_to_existing_list_at_lower(key, value, list, lower);
        }
    }
}


void StringIndex::insert_row_list(size_t ref, size_t offset, StringData index_data)
{
    REALM_ASSERT(!m_array->is_inner_bptree_node()); // only works in leaves

    // Create 4 byte index key
    key_type key = create_key(index_data, offset);

    // Get subnode table
    Allocator& alloc = m_array->get_alloc();
    Array values(alloc);
    get_child(*m_array, 0, values);
    REALM_ASSERT(m_array->size() == values.size() + 1);

    size_t ins_pos = values.lower_bound_int(key);
    if (ins_pos == values.size()) {
        // When key is outside current range, we can just add it
        values.add(key);
        m_array->add(ref);
        return;
    }

#ifdef REALM_DEBUG // LCOV_EXCL_START ignore debug code
    // Since we only use this for moving existing values to new
    // subindexes, there should never be an existing match.
    key_type k = key_type(values.get(ins_pos));
    REALM_ASSERT(k != key);
#endif // LCOV_EXCL_STOP ignore debug code

    // If key is not present we add it at the correct location
    values.insert(ins_pos, key);
    m_array->insert(ins_pos + 1, ref);
}


void StringIndex::TreeInsert(ObjKey obj_key, key_type key, size_t offset, StringData index_data, const Mixed& value)
{
    NodeChange nc = do_insert(obj_key, key, offset, index_data, value);
    switch (nc.type) {
        case NodeChange::change_None:
            return;
        case NodeChange::change_InsertBefore: {
            StringIndex new_node(inner_node_tag(), m_array->get_alloc());
            new_node.node_add_key(nc.ref1);
            new_node.node_add_key(get_ref());
            m_array->init_from_ref(new_node.get_ref());
            m_array->update_parent();
            return;
        }
        case NodeChange::change_InsertAfter: {
            StringIndex new_node(inner_node_tag(), m_array->get_alloc());
            new_node.node_add_key(get_ref());
            new_node.node_add_key(nc.ref1);
            m_array->init_from_ref(new_node.get_ref());
            m_array->update_parent();
            return;
        }
        case NodeChange::change_Split: {
            StringIndex new_node(inner_node_tag(), m_array->get_alloc());
            new_node.node_add_key(nc.ref1);
            new_node.node_add_key(nc.ref2);
            m_array->init_from_ref(new_node.get_ref());
            m_array->update_parent();
            return;
        }
    }
    REALM_ASSERT(false); // LCOV_EXCL_LINE; internal Realm error
}


StringIndex::NodeChange StringIndex::do_insert(ObjKey obj_key, key_type key, size_t offset, StringData index_data,
                                               const Mixed& value)
{
    Allocator& alloc = m_array->get_alloc();
    if (m_array->is_inner_bptree_node()) {
        // Get subnode table
        Array keys(alloc);
        get_child(*m_array, 0, keys);
        REALM_ASSERT(m_array->size() == keys.size() + 1);

        // Find the subnode containing the item
        size_t node_ndx = keys.lower_bound_int(key);
        if (node_ndx == keys.size()) {
            // node can never be empty, so try to fit in last item
            node_ndx = keys.size() - 1;
        }

        // Get sublist
        size_t refs_ndx = node_ndx + 1; // first entry in refs points to offsets
        ref_type ref = m_array->get_as_ref(refs_ndx);
        StringIndex target(ref, m_array.get(), refs_ndx, m_target_column, alloc);

        // Insert item
        NodeChange nc = target.do_insert(obj_key, key, offset, index_data, value);
        if (nc.type == NodeChange::change_None) {
            // update keys
            key_type last_key = target.get_last_key();
            keys.set(node_ndx, last_key);
            return NodeChange::change_None; // no new nodes
        }

        if (nc.type == NodeChange::change_InsertAfter) {
            ++node_ndx;
            ++refs_ndx;
        }

        // If there is room, just update node directly
        if (keys.size() < REALM_MAX_BPNODE_SIZE) {
            if (nc.type == NodeChange::change_Split) {
                node_insert_split(node_ndx, nc.ref2);
            }
            else {
                node_insert(node_ndx, nc.ref1); // ::INSERT_BEFORE/AFTER
            }
            return NodeChange::change_None;
        }

        // Else create new node
        StringIndex new_node(inner_node_tag(), alloc);
        if (nc.type == NodeChange::change_Split) {
            // update offset for left node
            key_type last_key = target.get_last_key();
            keys.set(node_ndx, last_key);

            new_node.node_add_key(nc.ref2);
            ++node_ndx;
            ++refs_ndx;
        }
        else {
            new_node.node_add_key(nc.ref1);
        }

        switch (node_ndx) {
            case 0: // insert before
                return NodeChange(NodeChange::change_InsertBefore, new_node.get_ref());
            case REALM_MAX_BPNODE_SIZE: // insert after
                if (nc.type == NodeChange::change_Split)
                    return NodeChange(NodeChange::change_Split, get_ref(), new_node.get_ref());
                return NodeChange(NodeChange::change_InsertAfter, new_node.get_ref());
            default: // split
                // Move items after split to new node
                size_t len = m_array->size();
                for (size_t i = refs_ndx; i < len; ++i) {
                    ref_type ref_i = m_array->get_as_ref(i);
                    new_node.node_add_key(ref_i);
                }
                keys.truncate(node_ndx);
                m_array->truncate(refs_ndx);
                return NodeChange(NodeChange::change_Split, get_ref(), new_node.get_ref());
        }
    }
    else {
        // Is there room in the list?
        Array old_keys(alloc);
        get_child(*m_array, 0, old_keys);
        const size_t old_offsets_size = old_keys.size();
        REALM_ASSERT_EX(m_array->size() == old_offsets_size + 1, m_array->size(), old_offsets_size + 1);

        bool noextend = old_offsets_size >= REALM_MAX_BPNODE_SIZE;

        // See if we can fit entry into current leaf
        // Works if there is room or it can join existing entries
        if (leaf_insert(obj_key, key, offset, index_data, value, noextend))
            return NodeChange::change_None;

        // Create new list for item (a leaf)
        StringIndex new_list(m_target_column, alloc);

        new_list.leaf_insert(obj_key, key, offset, index_data, value);

        size_t ndx = old_keys.lower_bound_int(key);

        // insert before
        if (ndx == 0)
            return NodeChange(NodeChange::change_InsertBefore, new_list.get_ref());

        // insert after
        if (ndx == old_offsets_size)
            return NodeChange(NodeChange::change_InsertAfter, new_list.get_ref());

        // split
        Array new_keys(alloc);
        get_child(*new_list.m_array, 0, new_keys);
        // Move items after split to new list
        for (size_t i = ndx; i < old_offsets_size; ++i) {
            int64_t v2 = old_keys.get(i);
            int64_t v3 = m_array->get(i + 1);

            new_keys.add(v2);
            new_list.m_array->add(v3);
        }
        old_keys.truncate(ndx);
        m_array->truncate(ndx + 1);

        return NodeChange(NodeChange::change_Split, get_ref(), new_list.get_ref());
    }
}


void StringIndex::node_insert_split(size_t ndx, size_t new_ref)
{
    REALM_ASSERT(m_array->is_inner_bptree_node());
    REALM_ASSERT(new_ref);

    Allocator& alloc = m_array->get_alloc();
    Array offsets(alloc);
    get_child(*m_array, 0, offsets);

    REALM_ASSERT(m_array->size() == offsets.size() + 1);
    REALM_ASSERT(ndx < offsets.size());
    REALM_ASSERT(offsets.size() < REALM_MAX_BPNODE_SIZE);

    // Get sublists
    size_t refs_ndx = ndx + 1; // first entry in refs points to offsets
    ref_type orig_ref = m_array->get_as_ref(refs_ndx);
    StringIndex orig_col(orig_ref, m_array.get(), refs_ndx, m_target_column, alloc);
    StringIndex new_col(new_ref, nullptr, 0, m_target_column, alloc);

    // Update original key
    key_type last_key = orig_col.get_last_key();
    offsets.set(ndx, last_key);

    // Insert new ref
    key_type new_key = new_col.get_last_key();
    offsets.insert(ndx + 1, new_key);
    m_array->insert(ndx + 2, new_ref);
}


void StringIndex::node_insert(size_t ndx, size_t ref)
{
    REALM_ASSERT(ref);
    REALM_ASSERT(m_array->is_inner_bptree_node());

    Allocator& alloc = m_array->get_alloc();
    Array offsets(alloc);
    get_child(*m_array, 0, offsets);
    REALM_ASSERT(m_array->size() == offsets.size() + 1);

    REALM_ASSERT(ndx <= offsets.size());
    REALM_ASSERT(offsets.size() < REALM_MAX_BPNODE_SIZE);

    StringIndex col(ref, nullptr, 0, m_target_column, alloc);
    key_type last_key = col.get_last_key();

    offsets.insert(ndx, last_key);
    m_array->insert(ndx + 1, ref);
}

bool StringIndex::leaf_insert(ObjKey obj_key, key_type key, size_t offset, StringData index_data, const Mixed& value,
                              bool noextend)
{
    REALM_ASSERT(!m_array->is_inner_bptree_node());
    if (offset >= s_max_offset && m_target_column.full_word()) {
        size_t len = value.get_string().size();
        size_t max = s_max_offset;
        throw LogicError(ErrorCodes::LimitExceeded,
                         util::format("String of length %1 exceeds maximum string length of %2.", len, max));
    }

    // Get subnode table
    Allocator& alloc = m_array->get_alloc();
    Array keys(alloc);
    get_child(*m_array, 0, keys);
    REALM_ASSERT(m_array->size() == keys.size() + 1);

    // If we are keeping the complete string in the index
    // we want to know if this is the last part
    bool is_at_string_end = offset + 4 >= index_data.size();

    size_t ins_pos = keys.lower_bound_int(key);
    size_t ins_pos_refs = ins_pos + 1; // first entry in refs points to offsets

    if (ins_pos == keys.size()) {
        if (noextend)
            return false;

        // When key is outside current range, we can just add it
        keys.add(key);
        if (!m_target_column.full_word() || is_at_string_end) {
            int64_t shifted = int64_t((uint64_t(obj_key.value) << 1) + 1); // shift to indicate literal
            m_array->add(shifted);
        }
        else {
            // create subindex for rest of string
            StringIndex subindex(m_target_column, m_array->get_alloc());
            subindex.insert_with_offset(obj_key, index_data, value, offset + 4);
            m_array->add(subindex.get_ref());
        }
        return true;
    }

    key_type k = key_type(keys.get(ins_pos));

    // If key is not present we add it at the correct location
    if (k != key) {
        if (noextend)
            return false;

        keys.insert(ins_pos, key);
        if (!m_target_column.full_word() || is_at_string_end) {
            int64_t shifted = int64_t((uint64_t(obj_key.value) << 1) + 1); // shift to indicate literal
            m_array->insert(ins_pos_refs, shifted);
        }
        else {
            // create subindex for rest of string
            StringIndex subindex(m_target_column, m_array->get_alloc());
            subindex.insert_with_offset(obj_key, index_data, value, offset + 4);
            m_array->insert(ins_pos_refs, subindex.get_ref());
        }
        return true;
    }

    // This leaf already has a slot for for the key

    uint64_t slot_value = uint64_t(m_array->get(ins_pos_refs));
    size_t suboffset = offset + s_index_key_length;

    // Single match (lowest bit set indicates literal row_ndx)
    if ((slot_value & 1) != 0) {
        ObjKey obj_key2 = ObjKey(int64_t(slot_value >> 1));
        Mixed v2 = m_target_column.full_word() ? reconstruct_string(offset, key, index_data) : get(obj_key2);
        if (v2 == value) {
            if (obj_key.value != obj_key2.value) {
                // Strings are equal but this is not a list.
                // Create a list and add both rows.

                // convert to list (in sorted order)
                Array row_list(alloc);
                row_list.create(Array::type_Normal); // Throws
                row_list.add(obj_key < obj_key2 ? obj_key.value : obj_key2.value);
                row_list.add(obj_key < obj_key2 ? obj_key2.value : obj_key.value);
                m_array->set(ins_pos_refs, row_list.get_ref());
            }
        }
        else {
            StringConversionBuffer buffer;
            auto index_data_2 = v2.get_index_data(buffer);
            if (index_data == index_data_2 || suboffset > s_max_offset) {
                // These strings have the same prefix up to this point but we
                // don't want to recurse further, create a list in sorted order.
                bool row_ndx_first = value < v2;
                Array row_list(alloc);
                row_list.create(Array::type_Normal); // Throws
                row_list.add(row_ndx_first ? obj_key.value : obj_key2.value);
                row_list.add(row_ndx_first ? obj_key2.value : obj_key.value);
                m_array->set(ins_pos_refs, row_list.get_ref());
            }
            else {
                // These strings have the same prefix up to this point but they
                // are actually not equal. Extend the tree recursivly until the
                // prefix of these strings is different.
                StringIndex subindex(m_target_column, m_array->get_alloc());
                subindex.insert_with_offset(obj_key2, index_data_2, v2, suboffset);
                subindex.insert_with_offset(obj_key, index_data, value, suboffset);
                // Join the string of SubIndices to the current position of m_array
                m_array->set(ins_pos_refs, subindex.get_ref());
            }
        }
        return true;
    }

    // If there already is a list of matches, we see if we fit there
    // or it has to be split into a subindex
    ref_type ref = ref_type(slot_value);
    char* header = alloc.translate(ref);
    if (!Array::get_context_flag_from_header(header)) {
        IntegerColumn sub(alloc, ref); // Throws
        sub.set_parent(m_array.get(), ins_pos_refs);

        IntegerColumn::const_iterator it_end = sub.cend();
        IntegerColumn::const_iterator lower = it_end;

        auto value_exists_in_list = [&]() {
            if (m_target_column.full_word()) {
                lower = sub.cbegin();
                return reconstruct_string(offset, key, index_data) == value.get_string();
            }
            SortedListComparator slc(m_target_column);
            lower = slc.find_start_of_unsorted(value, sub);

            if (lower != it_end) {
                Mixed lower_value = get(ObjKey(*lower));
                if (lower_value == value) {
                    return true;
                }
            }
            return false;
        };

        // If we found the value in this list, add the duplicate to the list.
        if (value_exists_in_list()) {
            insert_to_existing_list_at_lower(obj_key, value, sub, lower);
        }
        else {
            // If the list only stores duplicates we are free to branch and
            // and create a sub index with this existing list as one of the
            // leafs, but if the list doesn't only contain duplicates we
            // must respect that we store a common key prefix up to this
            // point and insert into the existing list.
            ObjKey key_of_any_dup = ObjKey(sub.get(0));
            StringConversionBuffer buffer;
            StringData index_data_2 = m_target_column.full_word() ? reconstruct_string(offset, key, index_data)
                                                                  : get(key_of_any_dup).get_index_data(buffer);
            if (index_data == index_data_2 || suboffset > s_max_offset) {
                insert_to_existing_list(obj_key, value, sub);
            }
            else {
#ifdef REALM_DEBUG
                bool contains_only_duplicates = true;
                if (!m_target_column.full_word() && sub.size() > 1) {
                    ObjKey first_key = ObjKey(sub.get(0));
                    ObjKey last_key = ObjKey(sub.back());
                    auto first = get(first_key);
                    auto last = get(last_key);
                    // Since the list is kept in sorted order, the first and
                    // last values will be the same only if the whole list is
                    // storing duplicate values.
                    if (first != last) {
                        contains_only_duplicates = false; // LCOV_EXCL_LINE
                    }
                }
                REALM_ASSERT_DEBUG(contains_only_duplicates);
#endif
                // The buffer is needed for when this is an integer index.
                StringIndex subindex(m_target_column, m_array->get_alloc());
                subindex.insert_row_list(sub.get_ref(), suboffset, index_data_2);
                subindex.insert_with_offset(obj_key, index_data, value, suboffset);
                m_array->set(ins_pos_refs, subindex.get_ref());
            }
        }
        return true;
    }

    // The key matches, but there is a subindex here so go down a level in the tree.
    StringIndex subindex(ref, m_array.get(), ins_pos_refs, m_target_column, alloc);
    subindex.insert_with_offset(obj_key, index_data, value, suboffset);

    return true;
}

Mixed StringIndex::get(ObjKey key) const
{
    return m_target_column.get_value(key);
}

void StringIndex::erase(ObjKey key)
{
    StringConversionBuffer buffer;
    if (m_target_column.full_word()) {
        if (m_target_column.tokenize()) {
            // This is a full text index
            auto index_data(get(key).get_index_data(buffer));
            auto words = Tokenizer::get_instance()->reset(std::string_view(index_data)).get_all_tokens();
            for (auto& w : words) {
                erase_string(key, w);
            }
        }
        else {
            // This is a list (of strings)
            erase_list(key, m_target_column.get_list(key));
        }
    }
    else {
        erase_string(key, get(key).get_index_data(buffer));
    }
}

void StringIndex::erase_list(ObjKey key, const Lst<String>& list)
{
    std::vector<StringData> strings;
    strings.reserve(list.size());
    for (auto& val : list) {
        strings.push_back(val);
    }

    std::sort(strings.begin(), strings.end());
    auto last = std::unique(strings.begin(), strings.end());
    for (auto it = strings.begin(); it != last; ++it) {
        erase_string(key, *it);
    }
}

namespace {
template <typename T>
void intersect(std::vector<ObjKey>& result, T& keys)
{
    if (result.empty()) {
        result.reserve(keys.size());
        for (auto k : keys) {
            result.emplace_back(k);
        }
    }
    else {
        auto it = result.begin();
        auto keep = it;
        auto m = keys.begin();

        // only keep intersection
        while (it != result.end() && m != keys.end()) {
            int64_t int_val = *m;
            if (it->value < int_val) {
                it++; // don't keep if match is not in new set
            }
            else if (it->value > int_val) {
                ++m; // ignore new matches
            }
            else {
                // Found both places - make sure it is kept
                if (keep < it)
                    *keep = *it;
                ++keep;
                ++it;
                ++m;
            }
        }
        if (keep != result.end()) {
            result.erase(keep, result.end());
        }
    }
}

struct FindResWrapper {
    InternalFindResult& res;
    IntegerColumn& indexes;
    size_t n = 0;
    size_t size()
    {
        return res.end_ndx - res.start_ndx;
    }
    auto begin()
    {
        return indexes.cbegin();
    }
    auto end()
    {
        return indexes.cend();
    }
};
} // namespace

void StringIndex::insert_bulk(const ArrayUnsigned* keys, uint64_t key_offset, size_t num_values, ArrayPayload& values)
{
    if (keys) {
        for (size_t i = 0; i < num_values; ++i) {
            ObjKey key(keys->get(i) + key_offset);
            insert(key, values.get_any(i));
        }
    }
    else {
        for (size_t i = 0; i < num_values; ++i) {
            ObjKey key(i + key_offset);
            insert(key, values.get_any(i));
        }
    }
}

void StringIndex::insert_bulk_list(const ArrayUnsigned* keys, uint64_t key_offset, size_t num_values,
                                   ArrayInteger& ref_array)
{
    auto get_obj_key = [&](size_t n) {
        if (keys) {
            return ObjKey(keys->get(n) + key_offset);
        }
        return ObjKey(n + key_offset);
    };
    for (size_t i = 0; i < num_values; ++i) {
        ObjKey key = get_obj_key(i);
        if (auto ref = to_ref(ref_array.get(i))) {
            BPlusTree<String> values(ref_array.get_alloc());
            values.init_from_ref(ref);
            values.for_all([&](const StringData& str) {
                insert(key, str);
            });
        }
    }
}


void StringIndex::find_all_fulltext(std::vector<ObjKey>& result, StringData value) const
{
    InternalFindResult res;
    REALM_ASSERT(result.empty());

    auto tokenizer = Tokenizer::get_instance();
    tokenizer->reset({value.data(), value.size()});
    auto [includes, excludes] = tokenizer->get_search_tokens();
    if (includes.empty()) {
        if (excludes.empty()) {
            throw InvalidArgument("Missing search token");
        }
        result = m_target_column.get_all_keys();
    }
    else {
        for (auto& token : includes) {
            if (token.back() == '*') {
                std::set<int64_t> keys;
                m_array->index_string_find_all_prefix(keys, StringData(token.data(), token.size() - 1));
                intersect(result, keys);
            }
            else {
                switch (find_all_no_copy(StringData{token}, res)) {
                    case FindRes_not_found:
                        result.clear();
                        break;
                    case FindRes_column: {
                        IntegerColumn indexes(m_array->get_alloc(), ref_type(res.payload));
                        FindResWrapper wrapper{res, indexes};
                        intersect(result, wrapper);
                        break;
                    }
                    case FindRes_single:
                        // merge in single res
                        if (result.empty()) {
                            result.emplace_back(res.payload);
                        }
                        else {
                            ObjKey key(res.payload);
                            auto pos = std::lower_bound(result.begin(), result.end(), key);
                            if (pos != result.end() && key == *pos) {
                                result.clear();
                                result.push_back(key);
                            }
                            else {
                                result.clear();
                            }
                        }
                        break;
                }
            }
            if (result.empty())
                return;
        }
    }

    for (auto& token : excludes) {
        if (token.back() == '*') {
            throw IllegalOperation("Exclude by prefix is not implemented");
        }
        if (result.empty())
            return;

        switch (find_all_no_copy(StringData{token}, res)) {
            case FindRes_not_found:
                // Nothing to exclude
                break;
            case FindRes_column: {
                IntegerColumn indexes(m_array->get_alloc(), ref_type(res.payload));

                auto it = result.begin();
                auto keep = it;
                size_t m = res.start_ndx;
                auto idx_val = indexes.get(m);

                while (it != result.end()) {
                    if (it->value < idx_val) {
                        // Not found in excludes
                        if (keep < it)
                            *keep = *it;
                        ++keep;
                        ++it;
                    }
                    else {
                        if (it->value == idx_val) {
                            // found in excludes - don't keep
                            ++it;
                        }
                        ++m;
                        idx_val = m < res.end_ndx ? indexes.get(m) : std::numeric_limits<int64_t>::max();
                    }
                }
                if (keep != result.end()) {
                    result.erase(keep, result.end());
                }
                break;
            }
            case FindRes_single: {
                // exclude single res
                ObjKey key(res.payload);
                auto pos = std::lower_bound(result.begin(), result.end(), key);
                if (pos != result.end() && key == *pos) {
                    result.erase(pos);
                }
                break;
            }
        }
    }
}


void StringIndex::clear()
{
    Array values(m_array->get_alloc());
    get_child(*m_array, 0, values);
    REALM_ASSERT(m_array->size() == values.size() + 1);

    values.clear();
    values.ensure_minimum_width(0x7FFFFFFF); // This ensures 31 bits plus a sign bit

    size_t size = 1;
    m_array->truncate_and_destroy_children(size); // Don't touch `values` array

    m_array->set_type(Array::type_HasRefs);
}


void StringIndex::do_delete(ObjKey obj_key, StringData index_data, size_t offset)
{
    Allocator& alloc = m_array->get_alloc();
    Array values(alloc);
    get_child(*m_array, 0, values);
    REALM_ASSERT(m_array->size() == values.size() + 1);

    // Create 4 byte index key
    key_type key = create_key(index_data, offset);

    const size_t pos = values.lower_bound_int(key);
    const size_t pos_refs = pos + 1; // first entry in refs points to offsets
    REALM_ASSERT(pos != values.size());

    if (m_array->is_inner_bptree_node()) {
        ref_type ref = m_array->get_as_ref(pos_refs);
        StringIndex node(ref, m_array.get(), pos_refs, m_target_column, alloc);
        node.do_delete(obj_key, index_data, offset);

        // Update the ref
        if (node.is_empty()) {
            values.erase(pos);
            m_array->erase(pos_refs);
            node.destroy();
        }
        else {
            key_type max_val = node.get_last_key();
            if (max_val != key_type(values.get(pos)))
                values.set(pos, max_val);
        }
    }
    else {
        uint64_t ref = m_array->get(pos_refs);
        if (ref & 1) {
            REALM_ASSERT(int64_t(ref >> 1) == obj_key.value);
            values.erase(pos);
            m_array->erase(pos_refs);
        }
        else {
            // A real ref either points to a list or a subindex
            char* header = alloc.translate(ref_type(ref));
            if (Array::get_context_flag_from_header(header)) {
                StringIndex subindex(ref_type(ref), m_array.get(), pos_refs, m_target_column, alloc);
                subindex.do_delete(obj_key, index_data, offset + s_index_key_length);

                if (subindex.is_empty()) {
                    values.erase(pos);
                    m_array->erase(pos_refs);
                    subindex.destroy();
                }
            }
            else {
                IntegerColumn sub(alloc, ref_type(ref)); // Throws
                sub.set_parent(m_array.get(), pos_refs);
                size_t r = sub.find_first(obj_key.value);
                size_t sub_size = sub.size(); // Slow
                REALM_ASSERT_EX(r != sub_size, r, sub_size);
                sub.erase(r);

                if (sub_size == 1) {
                    values.erase(pos);
                    m_array->erase(pos_refs);
                    sub.destroy();
                }
            }
        }
    }
}

void StringIndex::erase_string(ObjKey key, StringData value)
{
    do_delete(key, value, 0);

    // Collapse top nodes with single item
    while (m_array->is_inner_bptree_node()) {
        REALM_ASSERT(m_array->size() > 1); // node cannot be empty
        if (m_array->size() > 2)
            break;

        ref_type ref = m_array->get_as_ref(1);
        m_array->set(1, 1); // avoid destruction of the extracted ref
        m_array->destroy_deep();
        m_array->init_from_ref(ref);
        m_array->update_parent();
    }
}

namespace {

bool has_duplicate_values(const Array& node, const ClusterColumn& target_col) noexcept
{
    Allocator& alloc = node.get_alloc();
    Array child(alloc);
    size_t n = node.size();
    REALM_ASSERT(n >= 1);
    if (node.is_inner_bptree_node()) {
        // Inner node
        for (size_t i = 1; i < n; ++i) {
            ref_type ref = node.get_as_ref(i);
            child.init_from_ref(ref);
            if (has_duplicate_values(child, target_col))
                return true;
        }
        return false;
    }

    // Leaf node
    for (size_t i = 1; i < n; ++i) {
        int_fast64_t value = node.get(i);
        bool is_single_row_index = (value & 1) != 0;
        if (is_single_row_index)
            continue;

        ref_type ref = to_ref(value);
        child.init_from_ref(ref);

        bool is_subindex = child.get_context_flag();
        if (is_subindex) {
            if (has_duplicate_values(child, target_col))
                return true;
            continue;
        }

        // Child is root of B+-tree of row indexes
        IntegerColumn sub(alloc, ref);
        if (sub.size() > 1) {
            ObjKey first_key = ObjKey(sub.get(0));
            ObjKey last_key = ObjKey(sub.back());
            Mixed first = target_col.get_value(first_key);
            Mixed last = target_col.get_value(last_key);
            // Since the list is kept in sorted order, the first and
            // last values will be the same only if the whole list is
            // storing duplicate values.
            if (first == last) {
                return true;
            }
            // There may also be several short lists combined, so we need to
            // check each of these individually for duplicates.
            IntegerColumn::const_iterator it = sub.cbegin();
            IntegerColumn::const_iterator it_end = sub.cend();
            SortedListComparator slc(target_col);
            while (it != it_end) {
                Mixed it_data = target_col.get_value(ObjKey(*it));
                IntegerColumn::const_iterator next = slc.find_end_of_unsorted(it_data, sub, it);
                size_t count_of_value = next - it; // row index subtraction in `sub`
                if (count_of_value > 1) {
                    return true;
                }
                it = next;
            }
        }
    }

    return false;
}

} // anonymous namespace


bool StringIndex::has_duplicate_values() const noexcept
{
    return ::has_duplicate_values(*m_array, m_target_column);
}


bool StringIndex::is_empty() const
{
    return m_array->size() == 1; // first entry in refs points to offsets
}

void StringIndex::insert(ObjKey key, const Mixed& value)
{
    StringConversionBuffer buffer;
    constexpr size_t offset = 0; // First key from beginning of string

    if (this->m_target_column.tokenize()) {
        if (value.is_type(type_String)) {
            auto words = Tokenizer::get_instance()->reset(std::string_view(value.get<StringData>())).get_all_tokens();

            for (auto& word : words) {
                Mixed m(word);
                insert_with_offset(key, m.get_index_data(buffer), m, 0); // Throws
            }
        }
    }
    else {
        insert_with_offset(key, value.get_index_data(buffer), value, offset); // Throws
    }
}

void StringIndex::set(ObjKey key, const Mixed& new_value)
{
    StringConversionBuffer buffer;
    Mixed old_value = get(key);

    if (this->m_target_column.tokenize()) {
        auto tokenizer = Tokenizer::get_instance();
        StringData old_string = old_value.get_index_data(buffer);
        std::set<std::string> old_words;

        if (old_string.size() > 0) {
            tokenizer->reset({old_string.data(), old_string.size()});
            old_words = tokenizer->get_all_tokens();
        }
        std::set<std::string> new_words;
        if (new_value.is_type(type_String)) {
            new_words = tokenizer->reset(std::string_view(new_value.get<StringData>())).get_all_tokens();
        }

        auto w1 = old_words.begin();
        auto w2 = new_words.begin();

        // Do a diff, deleting words no longer present and
        // inserting new words
        while (w1 != old_words.end() && w2 != new_words.end()) {
            if (*w1 < *w2) {
                erase_string(key, *w1);
                ++w1;
            }
            else if (*w2 < *w1) {
                Mixed m(*w2);
                insert_with_offset(key, m.get_index_data(buffer), m, 0);
                ++w2;
            }
            else {
                ++w1;
                ++w2;
            }
        }
        while (w1 != old_words.end()) {
            erase_string(key, *w1);
            ++w1;
        }
        while (w2 != new_words.end()) {
            Mixed m(*w2);
            insert_with_offset(key, m.get_index_data(buffer), m, 0);

            ++w2;
        }
    }
    else {
        if (REALM_LIKELY(new_value != old_value)) {
            // We must erase this row first because erase uses find_first which
            // might find the duplicate if we insert before erasing.
            erase(key); // Throws

            auto index_data = new_value.get_index_data(buffer);
            insert_with_offset(key, index_data, new_value, 0); // Throws
        }
    }
}

void StringIndex::node_add_key(ref_type ref)
{
    REALM_ASSERT(ref);
    REALM_ASSERT(m_array->is_inner_bptree_node());

    Allocator& alloc = m_array->get_alloc();
    Array offsets(alloc);
    get_child(*m_array, 0, offsets);
    REALM_ASSERT(m_array->size() == offsets.size() + 1);
    REALM_ASSERT(offsets.size() < REALM_MAX_BPNODE_SIZE + 1);

    Array new_top(alloc);
    Array new_offsets(alloc);
    new_top.init_from_ref(ref);
    new_offsets.init_from_ref(new_top.get_as_ref(0));
    REALM_ASSERT(!new_offsets.is_empty());

    int64_t key = new_offsets.back();
    offsets.add(key);
    m_array->add(ref);
}

// Must return true if value of object(key) is less than 'b'.
bool SortedListComparator::operator()(int64_t key_value, const Mixed& b) // used in lower_bound
{
    Mixed a = m_column.get_value(ObjKey(key_value));
    if (a.is_null() && !b.is_null())
        return true;
    else if (b.is_null() && !a.is_null())
        return false;
    else if (a.is_null() && b.is_null())
        return false;
    return a.compare(b) < 0;
}


// Must return true if value of 'a' is less than value of object(key).
bool SortedListComparator::operator()(const Mixed& a, int64_t key_value) // used in upper_bound
{
    Mixed b = m_column.get_value(ObjKey(key_value));
    if (a.is_null() && !b.is_null())
        return true;
    else if (b.is_null() && !a.is_null())
        return false;
    else if (a.is_null() && b.is_null())
        return false;
    return a.compare(b) < 0;
}

// TODO: the next time the StringIndex is migrated (post version 23) and sorted
// properly in these lists replace this method with
// std::lower_bound(key_values.cbegin(), it_end, value, slc);
IntegerColumn::const_iterator SortedListComparator::find_start_of_unsorted(const Mixed& value,
                                                                           const IntegerColumn& key_values) const
{
    if (key_values.size() >= 2) {
        Mixed first = m_column.get_value(ObjKey(key_values.get(0)));
        Mixed last = m_column.get_value(ObjKey(key_values.get(key_values.size() - 1)));
        if (first == last) {
            if (value.compare(first) <= 0) {
                return key_values.cbegin();
            }
            else {
                return key_values.cend();
            }
        }
    }

    IntegerColumn::const_iterator it = key_values.cbegin();
    IntegerColumn::const_iterator end = key_values.cend();
    IntegerColumn::const_iterator first_greater = end;
    while (it != end) {
        Mixed val_it = m_column.get_value(ObjKey(*it));
        int cmp = val_it.compare(value);
        if (cmp == 0) {
            return it;
        }
        if (cmp > 0 && first_greater == end) {
            first_greater = it;
        }
        ++it;
    }
    return first_greater;
}

// TODO: same as above, change to std::upper_bound(lower, it_end, value, slc);
IntegerColumn::const_iterator SortedListComparator::find_end_of_unsorted(const Mixed& value,
                                                                         const IntegerColumn& key_values,
                                                                         IntegerColumn::const_iterator begin) const
{
    IntegerColumn::const_iterator end = key_values.cend();
    if (begin != end && end - begin > 0) {
        // optimization: check the last element first
        Mixed last = m_column.get_value(ObjKey(*(key_values.cend() - 1)));
        if (last == value) {
            return key_values.cend();
        }
    }
    while (begin != end) {
        Mixed val_it = m_column.get_value(ObjKey(*begin));
        if (value.compare(val_it) != 0) {
            return begin;
        }
        ++begin;
    }
    return end;
}

// LCOV_EXCL_START ignore debug functions
void StringIndex::verify() const
{
#ifdef REALM_DEBUG
    m_array->verify();

    Allocator& alloc = m_array->get_alloc();
    const size_t array_size = m_array->size();

    // Get first matching row for every key
    if (m_array->is_inner_bptree_node()) {
        for (size_t i = 1; i < array_size; ++i) {
            size_t ref = m_array->get_as_ref(i);
            StringIndex ndx(ref, nullptr, 0, m_target_column, alloc);
            ndx.verify();
        }
    }
    else {
        size_t column_size = m_target_column.size();
        for (size_t i = 1; i < array_size; ++i) {
            int64_t ref = m_array->get(i);

            // low bit set indicate literal ref (shifted)
            if (ref & 1) {
                size_t r = to_size_t((uint64_t(ref) >> 1));
                REALM_ASSERT_EX(r < column_size, r, column_size);
            }
            else {
                // A real ref either points to a list or a subindex
                char* header = alloc.translate(to_ref(ref));
                if (Array::get_context_flag_from_header(header)) {
                    StringIndex ndx(to_ref(ref), m_array.get(), i, m_target_column, alloc);
                    ndx.verify();
                }
                else {
                    IntegerColumn sub(alloc, to_ref(ref)); // Throws
                    IntegerColumn::const_iterator it = sub.cbegin();
                    IntegerColumn::const_iterator it_end = sub.cend();
                    SortedListComparator slc(m_target_column);
                    Mixed previous = get(ObjKey(*it));
                    size_t last_row = to_size_t(*it);

                    // Check that strings listed in sub are in sorted order
                    // and if there are duplicates, that the row numbers are
                    // sorted in the group of duplicates.
                    while (it != it_end) {
                        Mixed it_data = get(ObjKey(*it));
                        size_t it_row = to_size_t(*it);
                        REALM_ASSERT(previous <= it_data);
                        if (it != sub.cbegin() && previous == it_data) {
                            REALM_ASSERT_EX(it_row > last_row, it_row, last_row);
                        }
                        last_row = it_row;
                        previous = get(ObjKey(*it));
                        ++it;
                    }
                }
            }
        }
    }
// FIXME: Extend verification along the lines of IntegerColumn::verify().
#endif
}

#ifdef REALM_DEBUG

template <class T>
void StringIndex::verify_entries(const ClusterColumn& column) const
{
    std::vector<ObjKey> results;

    auto it = column.begin();
    auto end = column.end();
    auto col = column.get_column_key();
    while (it != end) {
        ObjKey key = it->get_key();
        T value = it->get<T>(col);

        find_all(results, value);

        auto ndx = find(results.begin(), results.end(), key);
        REALM_ASSERT(ndx != results.end());
        size_t found = count(value);
        REALM_ASSERT_EX(found >= 1, found);
        results.clear();
    }
}

namespace {

bool is_chars(uint64_t val)
{
    if (val == 0)
        return true;
    if (is_chars(val >> 8)) {
        char c = val & 0xFF;
        if (!c || std::isprint(c)) {
            return true;
        }
    }
    return false;
}

void out_char(std::ostream& out, uint64_t val)
{
    if (val) {
        out_char(out, val >> 8);
        char c = val & 0xFF;
        if (c && c != 'X') {
            out << c;
        }
    }
}

void out_hex(std::ostream& out, uint64_t val)
{
    if (is_chars(val)) {
        out_char(out, val);
    }
    else {
        out << int(val);
    }
}

} // namespace

void StringIndex::dump_node_structure(const Array& node, std::ostream& out, int level)
{
    int indent = level * 2;
    Allocator& alloc = node.get_alloc();
    Array subnode(alloc);

    size_t node_size = node.size();
    REALM_ASSERT(node_size >= 1);

    out << std::hex;

    bool node_is_leaf = !node.is_inner_bptree_node();
    if (node_is_leaf) {
        out << std::setw(indent) << ""
            << "Leaf (B+ tree) (ref: " << node.get_ref() << ")\n";
    }
    else {
        out << std::setw(indent) << ""
            << "Inner node (B+ tree) (ref: " << node.get_ref() << ")\n";
    }

    subnode.init_from_ref(to_ref(node.front()));
    out << std::setw(indent) << ""
        << "  Keys (keys_ref: " << subnode.get_ref() << ", ";
    if (subnode.is_empty()) {
        out << "no keys";
    }
    else {
        out << "keys: ";
        for (size_t i = 0; i != subnode.size(); ++i) {
            if (i != 0)
                out << ", ";
            out_hex(out, uint32_t(subnode.get(i)));
        }
    }
    out << ")\n";

    if (node_is_leaf) {
        for (size_t i = 1; i != node_size; ++i) {
            int_fast64_t value = node.get(i);
            bool is_single_row_index = (value & 1) != 0;
            if (is_single_row_index) {
                out << std::setw(indent) << ""
                    << "  Single row index (value: " << (value / 2) << ")\n";
                continue;
            }
            subnode.init_from_ref(to_ref(value));
            bool is_subindex = subnode.get_context_flag();
            if (is_subindex) {
                out << std::setw(indent) << ""
                    << "  Subindex\n";
                dump_node_structure(subnode, out, level + 2);
                continue;
            }
            IntegerColumn indexes(alloc, to_ref(value));
            out << std::setw(indent) << ""
                << "  List of row indexes\n";
            indexes.dump_values(out, level + 2);
        }
        return;
    }


    size_t num_children = node_size - 1;
    size_t child_ref_begin = 1;
    size_t child_ref_end = 1 + num_children;
    for (size_t i = child_ref_begin; i != child_ref_end; ++i) {
        subnode.init_from_ref(node.get_as_ref(i));
        dump_node_structure(subnode, out, level + 1);
    }
}

void StringIndex::print() const
{
    dump_node_structure(*m_array, std::cout, 0);
}

#endif // LCOV_EXCL_STOP ignore debug functions

realm / realm-core / thomas.goyne_478

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous