<commit_before>// Copyright 2010-2015 RethinkDB, all rights reserved.
#include "rdb_protocol/datum_stream.hpp"

#include <map>

#include "boost_utils.hpp"
#include "rdb_protocol/batching.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/func.hpp"
#include "rdb_protocol/term.hpp"
#include "rdb_protocol/val.hpp"
#include "utils.hpp"

#include "debug.hpp"

namespace ql {

read_mode_t up_to_date_read_mode(read_mode_t in) {
    switch (in) {
    case read_mode_t::MAJORITY: return in;
    case read_mode_t::SINGLE:   return in;
    case read_mode_t::OUTDATED: return read_mode_t::SINGLE;
    case read_mode_t::DEBUG_DIRECT:
        rfail_datum(base_exc_t::LOGIC,
                    "DEBUG_DIRECT is not a legal read mode for this operation "
                    "(an up-to-date read mode is required).");
    default: unreachable();
    }
}

key_range_t safe_universe() {
    return key_range_t(key_range_t::closed, store_key_t::min(),
                       key_range_t::open, store_key_t::max());
}

void debug_print(printf_buffer_t *buf, const range_state_t &rs) {
    const char *s;
    switch (rs) {
    case range_state_t::ACTIVE:    s = "ACTIVE"; break;
    case range_state_t::SATURATED: s = "SATURATED"; break;
    case range_state_t::EXHAUSTED: s = "EXHAUSTED"; break;
    default: unreachable();
    }
    buf->appendf("%s", s);
}

void debug_print(printf_buffer_t *buf, const hash_range_with_cache_t &hrwc) {
    buf->appendf("hash_range_with_cache_t(");
    debug_print(buf, hrwc.key_range);
    buf->appendf(", ");
    debug_print(buf, hrwc.cache);
    buf->appendf(", ");
    debug_print(buf, hrwc.state);
    buf->appendf(")\n");
}

void debug_print(printf_buffer_t *buf, const hash_ranges_t &hr) {
    debug_print(buf, hr.hash_ranges);
}

void debug_print(printf_buffer_t *buf, const active_ranges_t &ar) {
    debug_print(buf, ar.ranges);
}

bool hash_range_with_cache_t::totally_exhausted() const {
    return state == range_state_t::EXHAUSTED && cache.size() == 0;
}

range_state_t hash_ranges_t::state() const {
    bool seen_saturated = false;
    for (auto &&pair : hash_ranges) {
        switch (pair.second.state) {
        case range_state_t::ACTIVE: return range_state_t::ACTIVE;
        case range_state_t::SATURATED:
            seen_saturated = true;
            break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
    }
    return seen_saturated ? range_state_t::SATURATED : range_state_t::EXHAUSTED;
}

bool hash_ranges_t::totally_exhausted() const {
    for (auto &&pair : hash_ranges) {
        if (!pair.second.totally_exhausted()) return false;
    }
    return true;
}

bool active_ranges_t::totally_exhausted() const {
    for (auto &&pair : ranges) {
        if (!pair.second.totally_exhausted()) return false;
    }
    return true;
}

store_key_t truncate_and_get_left(active_ranges_t *ranges) {
    const store_key_t *smallest_left = &store_key_max;
    for (auto &&pair: ranges->ranges) {
        for (auto &&hash_pair : pair.second.hash_ranges) {
            if (hash_pair.second.cache.size() != 0) {
                // If we truncate the cache for a shard, it's always active.
                hash_pair.second.state = range_state_t::ACTIVE;
                // This should only ever be called when iterating left to right.
                hash_pair.second.key_range.left = hash_pair.second.cache[0].key;
                hash_pair.second.cache.clear();
            }
            if (hash_pair.second.key_range.left < *smallest_left) {
                smallest_left = &hash_pair.second.key_range.left;
            }
        }
    }
    return *smallest_left;
}

key_range_t active_ranges_to_range(const active_ranges_t &ranges) {
    store_key_t start = store_key_t::max();
    store_key_t end = store_key_t::min();
    bool seen_active = false;
    for (auto &&pair : ranges.ranges) {
        switch (pair.second.state()) {
        case range_state_t::ACTIVE:
            for (auto &&hash_pair : pair.second.hash_ranges) {
                start = std::min(start, hash_pair.second.key_range.left);
                end = std::max(end, hash_pair.second.key_range.right.key_or_max());
            }
            seen_active = true;
            break;
        case range_state_t::SATURATED:
            // If we didn't use any of the values we read last time, don't read
            // any more this time.
            break;
        case range_state_t::EXHAUSTED:
            // If there's no more data in a range then don't bother reading from it.
            break;
        default: unreachable();
        }
    }

    // We shouldn't get here unless there's at least one active range.
    r_sanity_check(seen_active);
    return key_range_t(key_range_t::closed, start, key_range_t::open, end);
}

boost::optional<std::map<region_t, store_key_t> > active_ranges_to_hints(
    sorting_t sorting, const boost::optional<active_ranges_t> &ranges) {

    if (!ranges) return boost::none;

    std::map<region_t, store_key_t> hints;
    for (auto &&pair : ranges->ranges) {
        switch (pair.second.state()) {
        case range_state_t::ACTIVE:
            for (auto &&hash_pair : pair.second.hash_ranges) {
                // If any of the hash shards in a range are active, we send
                // reads for all of them, because the assumption is that data is
                // randomly distributed across hash shards, which means the only
                // case where some are active and some are saturated is when
                // we're reading really small batches for some reason
                // (e.g. because of a sparse filter), in which case we still
                // want to read from all the hash shards at once.
                switch (hash_pair.second.state) {
                case range_state_t::ACTIVE: // fallthru
                case range_state_t::SATURATED:
                    hints[region_t(hash_pair.first.beg,
                                   hash_pair.first.end,
                                   pair.first)] = !reversed(sorting)
                        ? hash_pair.second.key_range.left
                        : hash_pair.second.key_range.right.key_or_max();
                    break;
                case range_state_t::EXHAUSTED: break;
                default: unreachable();
                }
            }
            break;
        case range_state_t::SATURATED: break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
    }
    r_sanity_check(hints.size() > 0);
    return std::move(hints);
}

enum class is_secondary_t { NO, YES };
active_ranges_t new_active_ranges(
    const stream_t &stream,
    key_range_t &&original_range,
    is_secondary_t is_secondary) {
    active_ranges_t ret;
    for (auto &&pair : stream.substreams) {
        std::map<hash_range_t, hash_range_with_cache_t> hash_ranges;
        ret.ranges[pair.first.inner]
           .hash_ranges[hash_range_t{pair.first.beg, pair.first.end}]
            = hash_range_with_cache_t{
                is_secondary == is_secondary_t::YES
                    ? std::move(original_range)
                    : pair.first.inner.intersection(original_range),
                raw_stream_t(),
                range_state_t::ACTIVE};
    }
    return ret;
}

class pseudoshard_t {
public:
    pseudoshard_t(sorting_t _sorting,
                  hash_range_with_cache_t *_cached,
                  keyed_stream_t *_fresh)
        : sorting(_sorting),
          cached(_cached),
          cached_index(0),
          fresh(_fresh),
          fresh_index(0),
          finished(false) {
        r_sanity_check(_cached != nullptr);
    }
    pseudoshard_t(pseudoshard_t &&) noexcept = default;
    pseudoshard_t &operator=(pseudoshard_t &&) = default;

    void finish() {
        r_sanity_check(!finished);
        switch (cached->state) {
        case range_state_t::ACTIVE:
            if (cached->cache.size() != 0) {
                if (cached_index == 0) {
                    cached->state = range_state_t::SATURATED;
                }
            } else if (fresh != nullptr && fresh->stream.size() != 0) {
                if (fresh_index == 0) {
                    cached->state = range_state_t::SATURATED;
                }
            } else {
                // No fresh values and no cached values means we should be exhausted.
                r_sanity_check(false);
            }
            break;
        case range_state_t::SATURATED:
            // A shard should never be saturated unless it has cached values.
            r_sanity_check(cached->cache.size() != 0);
            if (cached_index != 0) {
                cached->state = range_state_t::ACTIVE;
            }
            break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
        raw_stream_t new_cache;
        new_cache.reserve((cached->cache.size() - cached_index)
                          + (fresh ? (fresh->stream.size() - fresh_index) : 0));
        std::move(cached->cache.begin() + cached_index,
                  cached->cache.end(),
                  std::back_inserter(new_cache));
        if (fresh != nullptr) {
            std::move(fresh->stream.begin() + fresh_index,
                      fresh->stream.end(),
                      std::back_inserter(new_cache));
        }
        cached->cache = std::move(new_cache);
        finished = true;
    }

    const store_key_t *best_unpopped_key() const {
        r_sanity_check(!finished);
        if (cached_index < cached->cache.size()) {
            return &cached->cache[cached_index].key;
        } else if (fresh != nullptr && fresh_index < fresh->stream.size()) {
            return &fresh->stream[fresh_index].key;
        } else {
            if (!reversed(sorting)) {
                return cached->key_range.is_empty()
                    ? &store_key_max
                    : &cached->key_range.left;
            } else {
                return cached->key_range.is_empty()
                    ? &store_key_min
                    : &cached->key_range.right_or_max();
            }
        }
    }

    boost::optional<rget_item_t> pop() {
        r_sanity_check(!finished);
        if (cached_index < cached->cache.size()) {
            return std::move(cached->cache[cached_index++]);
        } else if (fresh != nullptr && fresh_index < fresh->stream.size()) {
            return std::move(fresh->stream[fresh_index++]);
        } else {
            return boost::none;
        }
    }
private:
    // We move out of `cached` and `fresh`, so we need to act like we have ownership.
    DISABLE_COPYING(pseudoshard_t);
    sorting_t sorting;
    hash_range_with_cache_t *cached;
    size_t cached_index;
    keyed_stream_t *fresh;
    size_t fresh_index;
    bool finished;
};

raw_stream_t rget_response_reader_t::unshard(
    sorting_t sorting,
    rget_read_response_t &&res) {

    grouped_t<stream_t> *gs = boost::get<grouped_t<stream_t> >(&res.result);
    r_sanity_check(gs != nullptr);
    auto stream = groups_to_batch(gs->get_underlying_map());
    if (!active_ranges) {
        active_ranges = new_active_ranges(
            stream, readgen->original_keyrange(res.reql_version),
            readgen->sindex_name() ? is_secondary_t::YES : is_secondary_t::NO);
        readgen->restrict_active_ranges(sorting, &*active_ranges);
        reql_version = res.reql_version;
    } else {
        r_sanity_check(res.reql_version == reql_version);
    }

    raw_stream_t ret;
    // Create the pseudoshards, which represent the cached, fresh, and
    // hypothetical `rget_item_t`s from the shards.  We also mark shards
    // exhausted in this step.
    std::vector<pseudoshard_t> pseudoshards;
    pseudoshards.reserve(active_ranges->ranges.size() * CPU_SHARDING_FACTOR);
    size_t n_active = 0, n_fresh = 0;
    for (auto &&pair : active_ranges->ranges) {
        for (auto &&hash_pair : pair.second.hash_ranges) {
            if (hash_pair.second.totally_exhausted()) continue;
            keyed_stream_t *fresh = nullptr;
            // Active shards need their bounds updated.
            if (hash_pair.second.state == range_state_t::ACTIVE) {<commit_msg>Fix unsharding problems that affected performance of small batches.<commit_after>// Copyright 2010-2015 RethinkDB, all rights reserved.
#include "rdb_protocol/datum_stream.hpp"

#include <map>

#include "boost_utils.hpp"
#include "rdb_protocol/batching.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/func.hpp"
#include "rdb_protocol/term.hpp"
#include "rdb_protocol/val.hpp"
#include "utils.hpp"

#include "debug.hpp"

namespace ql {

read_mode_t up_to_date_read_mode(read_mode_t in) {
    switch (in) {
    case read_mode_t::MAJORITY: return in;
    case read_mode_t::SINGLE:   return in;
    case read_mode_t::OUTDATED: return read_mode_t::SINGLE;
    case read_mode_t::DEBUG_DIRECT:
        rfail_datum(base_exc_t::LOGIC,
                    "DEBUG_DIRECT is not a legal read mode for this operation "
                    "(an up-to-date read mode is required).");
    default: unreachable();
    }
}

key_range_t safe_universe() {
    return key_range_t(key_range_t::closed, store_key_t::min(),
                       key_range_t::open, store_key_t::max());
}

void debug_print(printf_buffer_t *buf, const range_state_t &rs) {
    const char *s;
    switch (rs) {
    case range_state_t::ACTIVE:    s = "ACTIVE"; break;
    case range_state_t::SATURATED: s = "SATURATED"; break;
    case range_state_t::EXHAUSTED: s = "EXHAUSTED"; break;
    default: unreachable();
    }
    buf->appendf("%s", s);
}

void debug_print(printf_buffer_t *buf, const hash_range_with_cache_t &hrwc) {
    buf->appendf("hash_range_with_cache_t(");
    debug_print(buf, hrwc.key_range);
    buf->appendf(", ");
    debug_print(buf, hrwc.cache);
    buf->appendf(", ");
    debug_print(buf, hrwc.state);
    buf->appendf(")\n");
}

void debug_print(printf_buffer_t *buf, const hash_ranges_t &hr) {
    debug_print(buf, hr.hash_ranges);
}

void debug_print(printf_buffer_t *buf, const active_ranges_t &ar) {
    debug_print(buf, ar.ranges);
}

bool hash_range_with_cache_t::totally_exhausted() const {
    return state == range_state_t::EXHAUSTED && cache.size() == 0;
}

range_state_t hash_ranges_t::state() const {
    bool seen_saturated = false;
    for (auto &&pair : hash_ranges) {
        switch (pair.second.state) {
        case range_state_t::ACTIVE: return range_state_t::ACTIVE;
        case range_state_t::SATURATED:
            seen_saturated = true;
            break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
    }
    return seen_saturated ? range_state_t::SATURATED : range_state_t::EXHAUSTED;
}

bool hash_ranges_t::totally_exhausted() const {
    for (auto &&pair : hash_ranges) {
        if (!pair.second.totally_exhausted()) return false;
    }
    return true;
}

bool active_ranges_t::totally_exhausted() const {
    for (auto &&pair : ranges) {
        if (!pair.second.totally_exhausted()) return false;
    }
    return true;
}

store_key_t truncate_and_get_left(active_ranges_t *ranges) {
    const store_key_t *smallest_left = &store_key_max;
    for (auto &&pair: ranges->ranges) {
        for (auto &&hash_pair : pair.second.hash_ranges) {
            if (hash_pair.second.cache.size() != 0) {
                // If we truncate the cache for a shard, it's always active.
                hash_pair.second.state = range_state_t::ACTIVE;
                // This should only ever be called when iterating left to right.
                hash_pair.second.key_range.left = hash_pair.second.cache[0].key;
                hash_pair.second.cache.clear();
            }
            if (hash_pair.second.key_range.left < *smallest_left) {
                smallest_left = &hash_pair.second.key_range.left;
            }
        }
    }
    return *smallest_left;
}

key_range_t active_ranges_to_range(const active_ranges_t &ranges) {
    store_key_t start = store_key_t::max();
    store_key_t end = store_key_t::min();
    bool seen_active = false;
    for (auto &&pair : ranges.ranges) {
        switch (pair.second.state()) {
        case range_state_t::ACTIVE:
            for (auto &&hash_pair : pair.second.hash_ranges) {
                start = std::min(start, hash_pair.second.key_range.left);
                end = std::max(end, hash_pair.second.key_range.right.key_or_max());
            }
            seen_active = true;
            break;
        case range_state_t::SATURATED:
            // If we didn't use any of the values we read last time, don't read
            // any more this time.
            break;
        case range_state_t::EXHAUSTED:
            // If there's no more data in a range then don't bother reading from it.
            break;
        default: unreachable();
        }
    }

    // We shouldn't get here unless there's at least one active range.
    r_sanity_check(seen_active);
    return key_range_t(key_range_t::closed, start, key_range_t::open, end);
}

boost::optional<std::map<region_t, store_key_t> > active_ranges_to_hints(
    sorting_t sorting, const boost::optional<active_ranges_t> &ranges) {

    if (!ranges) return boost::none;

    std::map<region_t, store_key_t> hints;
    for (auto &&pair : ranges->ranges) {
        switch (pair.second.state()) {
        case range_state_t::ACTIVE:
            for (auto &&hash_pair : pair.second.hash_ranges) {
                // If any of the hash shards in a range are active, we send
                // reads for all of them, because the assumption is that data is
                // randomly distributed across hash shards, which means the only
                // case where some are active and some are saturated is when
                // we're reading really small batches for some reason
                // (e.g. because of a sparse filter), in which case we still
                // want to read from all the hash shards at once.
                switch (hash_pair.second.state) {
                case range_state_t::ACTIVE: // fallthru
                case range_state_t::SATURATED:
                    hints[region_t(hash_pair.first.beg,
                                   hash_pair.first.end,
                                   pair.first)] = !reversed(sorting)
                        ? hash_pair.second.key_range.left
                        : hash_pair.second.key_range.right.key_or_max();
                    break;
                case range_state_t::EXHAUSTED: break;
                default: unreachable();
                }
            }
            break;
        case range_state_t::SATURATED: break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
    }
    r_sanity_check(hints.size() > 0);
    return std::move(hints);
}

enum class is_secondary_t { NO, YES };
active_ranges_t new_active_ranges(
    const stream_t &stream,
    key_range_t &&original_range,
    is_secondary_t is_secondary) {
    active_ranges_t ret;
    for (auto &&pair : stream.substreams) {
        std::map<hash_range_t, hash_range_with_cache_t> hash_ranges;
        ret.ranges[pair.first.inner]
           .hash_ranges[hash_range_t{pair.first.beg, pair.first.end}]
            = hash_range_with_cache_t{
                is_secondary == is_secondary_t::YES
                    ? std::move(original_range)
                    : pair.first.inner.intersection(original_range),
                raw_stream_t(),
                range_state_t::ACTIVE};
    }
    return ret;
}

class pseudoshard_t {
public:
    pseudoshard_t(sorting_t _sorting,
                  hash_range_with_cache_t *_cached,
                  keyed_stream_t *_fresh)
        : sorting(_sorting),
          cached(_cached),
          cached_index(0),
          fresh(_fresh),
          fresh_index(0),
          finished(false) {
        r_sanity_check(_cached != nullptr);
    }
    pseudoshard_t(pseudoshard_t &&) noexcept = default;
    pseudoshard_t &operator=(pseudoshard_t &&) = default;

    void finish() {
        r_sanity_check(!finished);
        switch (cached->state) {
        case range_state_t::ACTIVE:
            if (cached->cache.size() != 0) {
                if (cached_index == 0) {
                    cached->state = range_state_t::SATURATED;
                }
            } else if (fresh != nullptr && fresh->stream.size() != 0) {
                if (fresh_index == 0) {
                    cached->state = range_state_t::SATURATED;
                }
            } else {
                // No fresh values and no cached values means we should be exhausted.
                r_sanity_check(false);
            }
            break;
        case range_state_t::SATURATED:
            // A shard should never be saturated unless it has cached values.
            r_sanity_check(cached->cache.size() != 0);
            if (cached_index != 0) {
                cached->state = range_state_t::ACTIVE;
            }
            break;
        case range_state_t::EXHAUSTED: break;
        default: unreachable();
        }
        if (fresh || cached_index > 0) {
            raw_stream_t new_cache;
            new_cache.reserve((cached->cache.size() - cached_index)
                              + (fresh ? (fresh->stream.size() - fresh_index) : 0));
            std::move(cached->cache.begin() + cached_index,
                      cached->cache.end(),
                      std::back_inserter(new_cache));
            if (fresh != nullptr) {
                std::move(fresh->stream.begin() + fresh_index,
                          fresh->stream.end(),
                          std::back_inserter(new_cache));
            }
            cached->cache = std::move(new_cache);
        }
        finished = true;
    }

    const store_key_t *best_unpopped_key() const {
        r_sanity_check(!finished);
        if (cached_index < cached->cache.size()) {
            return &cached->cache[cached_index].key;
        } else if (fresh != nullptr && fresh_index < fresh->stream.size()) {
            return &fresh->stream[fresh_index].key;
        } else {
            if (!reversed(sorting)) {
                return cached->key_range.is_empty()
                    ? &store_key_max
                    : &cached->key_range.left;
            } else {
                return cached->key_range.is_empty()
                    ? &store_key_min
                    : &cached->key_range.right_or_max();
            }
        }
    }

    boost::optional<rget_item_t> pop() {
        r_sanity_check(!finished);
        if (cached_index < cached->cache.size()) {
            return std::move(cached->cache[cached_index++]);
        } else if (fresh != nullptr && fresh_index < fresh->stream.size()) {
            return std::move(fresh->stream[fresh_index++]);
        } else {
            return boost::none;
        }
    }
private:
    // We move out of `cached` and `fresh`, so we need to act like we have ownership.
    DISABLE_COPYING(pseudoshard_t);
    sorting_t sorting;
    hash_range_with_cache_t *cached;
    size_t cached_index;
    keyed_stream_t *fresh;
    size_t fresh_index;
    bool finished;
};

raw_stream_t rget_response_reader_t::unshard(
    sorting_t sorting,
    rget_read_response_t &&res) {

    grouped_t<stream_t> *gs = boost::get<grouped_t<stream_t> >(&res.result);
    r_sanity_check(gs != nullptr);
    auto stream = groups_to_batch(gs->get_underlying_map());
    if (!active_ranges) {
        active_ranges = new_active_ranges(
            stream, readgen->original_keyrange(res.reql_version),
            readgen->sindex_name() ? is_secondary_t::YES : is_secondary_t::NO);
        readgen->restrict_active_ranges(sorting, &*active_ranges);
        reql_version = res.reql_version;
    } else {
        r_sanity_check(res.reql_version == reql_version);
    }

    raw_stream_t ret;
    // Create the pseudoshards, which represent the cached, fresh, and
    // hypothetical `rget_item_t`s from the shards.  We also mark shards
    // exhausted in this step.
    std::vector<pseudoshard_t> pseudoshards;
    pseudoshards.reserve(active_ranges->ranges.size() * CPU_SHARDING_FACTOR);
    size_t n_active = 0, n_fresh = 0;
    for (auto &&pair : active_ranges->ranges) {
        bool range_active = pair.second.state() == range_state_t::ACTIVE;
        for (auto &&hash_pair : pair.second.hash_ranges) {
            if (hash_pair.second.totally_exhausted()) continue;
            keyed_stream_t *fresh = nullptr;
            // Active shards need their bounds updated.
            if (range_active) {
{'fraction_non_alphanumeric': 0.1108400422867366, 'fraction_numerical': 0.0019110352118402861, 'mean_word_length': 2.231082501313715, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 106, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6866238', 'n_tokens_mistral': 7951, 'n_tokens_neox': 7702, 'n_words': 1595}