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

#include <algorithm>

#include "errors.hpp"
#include <boost/bind.hpp>
#include <boost/make_shared.hpp>

#include "arch/io/disk.hpp"
#include "btree/erase_range.hpp"
#include "btree/parallel_traversal.hpp"
#include "btree/slice.hpp"
#include "btree/superblock.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_watchable.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/vector_stream.hpp"
#include "protob/protob.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/transform_visitors.hpp"
#include "rdb_protocol/pb_utils.hpp"
#include "rdb_protocol/term_walker.hpp"
#include "rpc/semilattice/view/field.hpp"
#include "rpc/semilattice/watchable.hpp"
#include "serializer/config.hpp"

// Ignore clang errors for ql macros
#pragma GCC diagnostic ignored "-Wshadow"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_t;
typedef rdb_protocol_details::range_key_tester_t range_key_tester_t;

typedef rdb_protocol_t::context_t context_t;

typedef rdb_protocol_t::store_t store_t;
typedef rdb_protocol_t::region_t region_t;

typedef rdb_protocol_t::read_t read_t;
typedef rdb_protocol_t::read_response_t read_response_t;

typedef rdb_protocol_t::point_read_t point_read_t;
typedef rdb_protocol_t::point_read_response_t point_read_response_t;

typedef rdb_protocol_t::rget_read_t rget_read_t;
typedef rdb_protocol_t::rget_read_response_t rget_read_response_t;

typedef rdb_protocol_t::distribution_read_t distribution_read_t;
typedef rdb_protocol_t::distribution_read_response_t distribution_read_response_t;

typedef rdb_protocol_t::sindex_list_t sindex_list_t;
typedef rdb_protocol_t::sindex_list_response_t sindex_list_response_t;

typedef rdb_protocol_t::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_t;

typedef rdb_protocol_t::point_replace_t point_replace_t;
typedef rdb_protocol_t::point_replace_response_t point_replace_response_t;

typedef rdb_protocol_t::batched_replaces_t batched_replaces_t;

typedef rdb_protocol_t::point_write_t point_write_t;
typedef rdb_protocol_t::point_write_response_t point_write_response_t;

typedef rdb_protocol_t::point_delete_t point_delete_t;
typedef rdb_protocol_t::point_delete_response_t point_delete_response_t;

typedef rdb_protocol_t::sindex_create_t sindex_create_t;
typedef rdb_protocol_t::sindex_create_response_t sindex_create_response_t;

typedef rdb_protocol_t::sindex_drop_t sindex_drop_t;
typedef rdb_protocol_t::sindex_drop_response_t sindex_drop_response_t;

typedef rdb_protocol_t::backfill_chunk_t backfill_chunk_t;

typedef rdb_protocol_t::backfill_progress_t backfill_progress_t;

typedef rdb_protocol_t::rget_read_response_t::stream_t stream_t;
typedef rdb_protocol_t::rget_read_response_t::groups_t groups_t;
typedef rdb_protocol_t::rget_read_response_t::atom_t atom_t;
typedef rdb_protocol_t::rget_read_response_t::length_t length_t;
typedef rdb_protocol_t::rget_read_response_t::inserted_t inserted_t;

typedef btree_store_t<rdb_protocol_t>::sindex_access_vector_t sindex_access_vector_t;

const std::string rdb_protocol_t::protocol_name("rdb");

RDB_IMPL_PROTOB_SERIALIZABLE(Term);
RDB_IMPL_PROTOB_SERIALIZABLE(Datum);


namespace rdb_protocol_details {

RDB_IMPL_SERIALIZABLE_3(backfill_atom_t, key, value, recency);
RDB_IMPL_SERIALIZABLE_3(transform_atom_t, variant, scopes, backtrace);
RDB_IMPL_SERIALIZABLE_3(terminal_t, variant, scopes, backtrace);

void post_construct_and_drain_queue(
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue,
        auto_drainer_t::lock_t lock)
    THROWS_NOTHING;
/* Creates a queue of operations for the sindex, runs a post construction for
 * the data already in the btree and finally drains the queue. */
void bring_sindexes_up_to_date(
        const std::set<std::string> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        buf_lock_t *sindex_block,
        transaction_t *txn)
    THROWS_NOTHING
{
    /* We register our modification queue here. An important point about
     * correctness here: we've held the superblock this whole time and will
     * continue to do so until the call to post_construct_secondary_indexes
     * begins a parallel traversal which releases the superblock. This
     * serves to make sure that every changes which we don't learn about in
     * the parallel traversal we do learn about from the mod queue. */
    uuid_u post_construct_id = generate_uuid();

    boost::shared_ptr<internal_disk_backed_queue_t> mod_queue(
            new internal_disk_backed_queue_t(
                store->io_backender_,
                serializer_filepath_t(store->base_path_, "post_construction_" + uuid_to_str(post_construct_id)),
                &store->perfmon_collection));

    {
        mutex_t::acq_t acq;
        store->lock_sindex_queue(sindex_block, &acq);
        store->register_sindex_queue(mod_queue.get(), &acq);
    }

    std::map<std::string, secondary_index_t> sindexes;
    store->get_sindexes(sindex_block, txn, &sindexes);
    std::set<uuid_u> sindexes_to_bring_up_to_date_uuid;

    for (auto it = sindexes_to_bring_up_to_date.begin(); 
         it != sindexes_to_bring_up_to_date.end(); ++it) {
        guarantee(std_contains(sindexes, *it));
        sindexes_to_bring_up_to_date_uuid.insert(sindexes[*it].id);
    }

    coro_t::spawn_sometime(boost::bind(
                &post_construct_and_drain_queue,
                sindexes_to_bring_up_to_date_uuid,
                store,
                mod_queue,
                auto_drainer_t::lock_t(&store->drainer)));
}

/* This function is really part of the logic of bring_sindexes_up_to_date
 * however it needs to be in a seperate function so that it can be spawned in a
 * coro. */
void post_construct_and_drain_queue(
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue,
        auto_drainer_t::lock_t lock)
    THROWS_NOTHING
{
    try {
        post_construct_secondary_indexes(store, sindexes_to_bring_up_to_date, lock.get_drain_signal());

        /* Drain the queue. */

        int previous_size = mod_queue->size();
        while (!lock.get_drain_signal()->is_pulsed()) {
            write_token_pair_t token_pair;
            store->new_write_token_pair(&token_pair);

            scoped_ptr_t<transaction_t> queue_txn;
            scoped_ptr_t<real_superblock_t> queue_superblock;

            // We don't need hard durability here, because a secondary index just gets rebuilt
            // if the server dies while it's partially constructed.
            store->acquire_superblock_for_write(
                rwi_write,
                repli_timestamp_t::distant_past,
                2,
                WRITE_DURABILITY_SOFT,
                &token_pair.main_write_token,
                &queue_txn,
                &queue_superblock,
                lock.get_drain_signal());

            scoped_ptr_t<buf_lock_t> queue_sindex_block;
            store->acquire_sindex_block_for_write(
                &token_pair,
                queue_txn.get(),
                &queue_sindex_block,
                queue_superblock->get_sindex_block_id(),
                lock.get_drain_signal());

            sindex_access_vector_t sindexes;
            store->acquire_sindex_superblocks_for_write(
                    sindexes_to_bring_up_to_date,
                    queue_sindex_block.get(),
                    queue_txn.get(),
                    &sindexes);

            if (sindexes.empty()) {
                break;
            }

            mutex_t::acq_t acq;
            store->lock_sindex_queue(queue_sindex_block.get(), &acq);

            while (mod_queue->size() >= previous_size &&
                   mod_queue->size() > 0) {
                std::vector<char> data_vec;
                mod_queue->pop(&data_vec);
                vector_read_stream_t read_stream(&data_vec);

                rdb_modification_report_t mod_report;
                int ser_res = deserialize(&read_stream, &mod_report);
                guarantee_err(ser_res == 0, "corruption in disk-backed queue");

                rdb_update_sindexes(sindexes, &mod_report, queue_txn.get());
            }

            previous_size = mod_queue->size();

            if (mod_queue->size() == 0) {
                for (auto it = sindexes_to_bring_up_to_date.begin();
                     it != sindexes_to_bring_up_to_date.end(); ++it) {
                    store->mark_index_up_to_date(*it, queue_txn.get(), queue_sindex_block.get());
                }
                store->deregister_sindex_queue(mod_queue.get(), &acq);
                return;
            }
        }
    } catch (const interrupted_exc_t &) {
        // We were interrupted so we just exit. Sindex post construct is in an
        // indeterminate state and will be cleaned up at a later point.
    }

    {
        /* If we get here it's either because we were interrupted or the
         * sindexes we were post constructiing were deleted. Either way we need
         * to clean up the queue. */
        write_token_pair_t token_pair;
        store->new_write_token_pair(&token_pair);

        scoped_ptr_t<transaction_t> queue_txn;
        scoped_ptr_t<real_superblock_t> queue_superblock;

        store->acquire_superblock_for_write(
            rwi_write,
            repli_timestamp_t::distant_past,
            2,
            WRITE_DURABILITY_HARD,
            &token_pair.main_write_token,
            &queue_txn,
            &queue_superblock,
            lock.get_drain_signal());

        scoped_ptr_t<buf_lock_t> queue_sindex_block;
        store->acquire_sindex_block_for_write(
            &token_pair,
            queue_txn.get(),
            &queue_sindex_block,
            queue_superblock->get_sindex_block_id(),
            lock.get_drain_signal());

        mutex_t::acq_t acq;
        store->lock_sindex_queue(queue_sindex_block.get(), &acq);
        store->deregister_sindex_queue(mod_queue.get(), &acq);
    }
}


bool range_key_tester_t::key_should_be_erased(const btree_key_t *key) {
    uint64_t h = hash_region_hasher(key->contents, key->size);
    return delete_range->beg <= h && h < delete_range->end
        && delete_range->inner.contains_key(key->contents, key->size);
}

}  // namespace rdb_protocol_details

rdb_protocol_t::context_t::context_t()
    : pool_group(NULL), ns_repo(NULL),
    cross_thread_namespace_watchables(get_num_threads()),
    cross_thread_database_watchables(get_num_threads()),
    directory_read_manager(NULL),
    signals(get_num_threads())
{ }

rdb_protocol_t::context_t::context_t(
    extproc::pool_group_t *_pool_group,
    namespace_repo_t<rdb_protocol_t> *_ns_repo,
    boost::shared_ptr<semilattice_readwrite_view_t<cluster_semilattice_metadata_t> >
        _semilattice_metadata,
    directory_read_manager_t<cluster_directory_metadata_t>
        *_directory_read_manager,
    machine_id_t _machine_id)
    : pool_group(_pool_group), ns_repo(_ns_repo),
      cross_thread_namespace_watchables(get_num_threads()),
      cross_thread_database_watchables(get_num_threads()),
      semilattice_metadata(_semilattice_metadata),
      directory_read_manager(_directory_read_manager),
      signals(get_num_threads()),
      machine_id(_machine_id)
{
    for (int thread = 0; thread < get_num_threads(); ++thread) {
        cross_thread_namespace_watchables[thread].init(new cross_thread_watchable_variable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > >(
                                                    clone_ptr_t<semilattice_watchable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > > >
                                                        (new semilattice_watchable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > >(
                                                            metadata_field(&cluster_semilattice_metadata_t::rdb_namespaces, _semilattice_metadata))), thread));

        cross_thread_database_watchables[thread].init(new cross_thread_watchable_variable_t<databases_semilattice_metadata_t>(
                                                    clone_ptr_t<semilattice_watchable_t<databases_semilattice_metadata_t> >
                                                        (new semilattice_watchable_t<databases_semilattice_metadata_t>(
                                                            metadata_field(&cluster_semilattice_metadata_t::databases, _semilattice_metadata))), thread));

        signals[thread].init(new cross_thread_signal_t(&interruptor, thread));
    }
}

rdb_protocol_t::context_t::~context_t() { }

// Construct a region containing only the specified key
region_t rdb_protocol_t::monokey_region(const store_key_t &k) {
    uint64_t h = hash_region_hasher(k.contents(), k.size());
    return region_t(h, h + 1, key_range_t(key_range_t::closed, k, key_range_t::closed, k));
}

key_range_t rdb_protocol_t::sindex_key_range(const store_key_t &start, const store_key_t &end) {
    std::string start_key_str(key_to_unescaped_str(start));
    std::string end_key_str(key_to_unescaped_str(end));
    store_key_t end_key;

    // Need to make the next largest store_key_t without making the key longer
    while (end_key_str.length() > 0 &&
           end_key_str[end_key_str.length() - 1] == static_cast<char>(255)) {
        end_key_str.erase(end_key_str.length() - 1);
    }

    if (end_key_str.length() == 0) {
        end_key = store_key_t::max();
    } else {
        ++end_key_str[end_key_str.length() - 1];
        end_key = store_key_t(end_key_str);
    }

    return key_range_t(key_range_t::closed, store_key_t(start_key_str),
                       key_range_t::open, end_key);
}

/* read_t::get_region implementation */
struct rdb_r_get_region_visitor : public boost::static_visitor<region_t> {
    region_t operator()(const point_read_t &pr) const {
        return rdb_protocol_t::monokey_region(pr.key);
    }

    region_t operator()(const rget_read_t &rg) const {
        return rg.region;
    }

    region_t operator()(const distribution_read_t &dg) const {
        return dg.region;
    }

    region_t operator()(UNUSED const sindex_list_t &sl) const {
        return rdb_protocol_t::monokey_region(store_key_t());
    }
};

region_t read_t::get_region() const THROWS_NOTHING {
    return boost::apply_visitor(rdb_r_get_region_visitor(), read);
}

struct rdb_r_shard_visitor : public boost::static_visitor<read_t> {
    explicit rdb_r_shard_visitor(const region_t &_region)
        : region(_region)
    { }

    read_t operator()(const point_read_t &pr) const {
        rassert(rdb_protocol_t::monokey_region(pr.key) == region);
        return read_t(pr);
    }

    read_t operator()(const rget_read_t &rg) const {
        rassert(region_is_superset(rg.region, region));
        rget_read_t _rg(rg);
        _rg.region = region;
        return read_t(_rg);
    }

    read_t operator()(const distribution_read_t &dg) const {
        rassert(region_is_superset(dg.region, region));
        distribution_read_t _dg(dg);
        _dg.region = region;
        return read_t(_dg);
    }

    read_t operator()(const sindex_list_t &sl) const {
        return read_t(sl);
    }

    const region_t &region;
};

read_t read_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(rdb_r_shard_visitor(region), read);
}

/* read_t::unshard implementation */
bool read_response_cmp(const read_response_t &l, const read_response_t &r) {
    const rget_read_response_t *lr = boost::get<rget_read_response_t>(&l.response);
    guarantee(lr);
    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&r.response);
    guarantee(rr);
    return lr->key_range < rr->key_range;
}

/* A visitor to handle this unsharding process for us. */

class distribution_read_response_less_t {
public:
    bool operator()(const distribution_read_response_t& x, const distribution_read_response_t& y) {
        if (x.region.inner == y.region.inner) {
            return x.region < y.region;
        } else {
            return x.region.inner < y.region.inner;
        }
    }
};

// Scale the distribution down by combining ranges to fit it within the limit of
// the query
void scale_down_distribution(size_t result_limit, std::map<store_key_t, int64_t> *key_counts) {
    guarantee(result_limit > 0);
    const size_t combine = (key_counts->size() / result_limit); // Combine this many other ranges into the previous range
    for (std::map<store_key_t, int64_t>::iterator it = key_counts->begin(); it != key_counts->end(); ) {
        std::map<store_key_t, int64_t>::iterator next = it;
        ++next;
        for (size_t i = 0; i < combine && next != key_counts->end(); ++i) {
            it->second += next->second;
            std::map<store_key_t, int64_t>::iterator tmp = next;
            ++next;
            key_counts->erase(tmp);
        }
        it = next;
    }
}

class rdb_r_unshard_visitor_t : public boost::static_visitor<void> {
public:
    rdb_r_unshard_visitor_t(const read_response_t *_responses,
                            size_t _count,
                            read_response_t *_response_out,
                            rdb_protocol_t::context_t *ctx,
                            signal_t *interruptor)
        : responses(_responses), count(_count), response_out(_response_out),
          ql_env(ctx->pool_group,
                 ctx->ns_repo,
                 ctx->cross_thread_namespace_watchables[get_thread_id()].get()
                     ->get_watchable(),
                 ctx->cross_thread_database_watchables[get_thread_id()].get()
                     ->get_watchable(),
                 ctx->semilattice_metadata,
                 NULL,
                 boost::make_shared<js::runner_t>(),
                 interruptor,
                 ctx->machine_id,
                 std::map<std::string, ql::wire_func_t>())
    { }

    void operator()(const point_read_t &) {
        guarantee(count == 1);
        guarantee(NULL != boost::get<point_read_response_t>(&responses[0].response));
        *response_out = responses[0];
    }

    void operator()(const rget_read_t &rg) {
        response_out->response = rget_read_response_t();
        rget_read_response_t *rg_response = boost::get<rget_read_response_t>(&response_out->response);
        rg_response->truncated = false;
        rg_response->key_range = read_t(rg).get_region().inner;
        rg_response->last_considered_key = read_t(rg).get_region().inner.left;

        try {
            /* First check to see if any of the responses we're unsharding threw. */
            for (size_t i = 0; i < count; ++i) {
                // TODO: we're ignoring the limit when recombining.
                const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                guarantee(rr != NULL);

                if (const runtime_exc_t *e = boost::get<runtime_exc_t>(&rr->result)) {
                    throw *e;
                } else if (const ql::exc_t *e2 = boost::get<ql::exc_t>(&rr->result)) {
                    throw *e2;
                } else if (const ql::datum_exc_t *e3 = boost::get<ql::datum_exc_t>(&rr->result)) {
                    throw *e3;
                }
            }

            if (!rg.terminal) {
                //A vanilla range get
                //First we need to determine the cutoff key:
                rg_response->last_considered_key = store_key_t::max();
                for (size_t i = 0; i < count; ++i) {
                    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                    guarantee(rr != NULL);

                    if (rg_response->last_considered_key > rr->last_considered_key && rr->truncated) {
                        rg_response->last_considered_key = rr->last_considered_key;
                    }
                }

                rg_response->result = stream_t();
                stream_t *res_stream = boost::get<stream_t>(&rg_response->result);
                for (size_t i = 0; i < count; ++i) {
                    // TODO: we're ignoring the limit when recombining.
                    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                    guarantee(rr != NULL);

                    const stream_t *stream = boost::get<stream_t>(&(rr->result));

                    for (stream_t::const_iterator it = stream->begin(); it != stream->end(); ++it) {
                        if (it->first <= rg_response->last_considered_key) {
                            res_stream->push_back(*it);
                        }
                    }

                    rg_response->truncated = rg_response->truncated || rr->truncated;
                }
            } else {
                try {
                    if (const ql::reduce_wire_func_t *reduce_func =
                            boost::get<ql::reduce_wire_func_t>(&rg.terminal->variant)) {
                        ql::reduce_wire_func_t local_reduce_func = *reduce_func;
                        rg_response->result = rget_read_response_t::empty_t();
                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            ql::wire_datum_t *lhs = boost::get<ql::wire_datum_t>(&rg_response->result);
                            const ql::wire_datum_t *rhs = boost::get<ql::wire_datum_t>(&(_rr->result));
                            if (!rhs) {
                                guarantee(boost::get<rget_read_response_t::empty_t>(&(_rr->result)));
                                continue;
                            } else {
                                ql::wire_datum_t local_rhs = *rhs;
                                if (lhs) {
                                    const ql::datum_t *reduced_val =
                                        local_reduce_func.
                                            compile(&ql_env)->
                                                call(lhs->compile(&ql_env), local_rhs.compile(&ql_env))->
                                                    as_datum();
                                    rg_response->result = ql::wire_datum_t(reduced_val);
                                } else {
                                    guarantee(boost::get<rget_read_response_t::empty_t>(&rg_response->result));
                                    rg_response->result = _rr->result;
                                }
                            }
                        }
                        ql::wire_datum_t *final_val = boost::get<ql::wire_datum_t>(&rg_response->result);
                        if (final_val) {
                            final_val->finalize();
                        }
                    } else if (boost::get<ql::count_wire_func_t>(&rg.terminal->variant)) {
                        rg_response->result =
                            ql::wire_datum_t(ql_env.add_ptr(new ql::datum_t(0.0)));

                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            ql::wire_datum_t *lhs = boost::get<ql::wire_datum_t>(&rg_response->result);
                            const ql::wire_datum_t *rhs = boost::get<ql::wire_datum_t>(&(_rr->result));
                            ql::wire_datum_t local_rhs = *rhs;

                            const ql::datum_t *sum =
                                ql_env.add_ptr(new ql::datum_t(
                                                   lhs->compile(&ql_env)->as_num() +
                                                   local_rhs.compile(&ql_env)->as_num()));
                            rg_response->result = ql::wire_datum_t(sum);
                        }
                        boost::get<ql::wire_datum_t>(rg_response->result).finalize();
                    } else if (const ql::gmr_wire_func_t *gmr_func =
                            boost::get<ql::gmr_wire_func_t>(&rg.terminal->variant)) {
                        ql::gmr_wire_func_t local_gmr_func = *gmr_func;
                        rg_response->result = ql::wire_datum_map_t();
                        ql::wire_datum_map_t *map =
                            boost::get<ql::wire_datum_map_t>(&rg_response->result);

                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            const ql::wire_datum_map_t *rhs =
                                boost::get<ql::wire_datum_map_t>(&(_rr->result));
                            r_sanity_check(rhs);
                            ql::wire_datum_map_t local_rhs = *rhs;
                            local_rhs.compile(&ql_env);

                            const ql::datum_t *rhs_arr = local_rhs.to_arr(&ql_env);
                            for (size_t f = 0; f < rhs_arr->size(); ++f) {
                                const ql::datum_t *key = rhs_arr->get(f)->get("group");
                                const ql::datum_t *val = rhs_arr->get(f)->get("reduction");
                                if (!map->has(key)) {
                                    map->set(key, val);
                                } else {
                                    ql::func_t *r = local_gmr_func.compile_reduce(&ql_env);
                                    map->set(key, r->call(map->get(key), val)->as_datum());
                                }
                            }
                        }
                        boost::get<ql::wire_datum_map_t>(rg_response->result).finalize();
                    } else {
                        unreachable();
                    }
                } catch (const ql::datum_exc_t &e) {
                    /* Evaluation threw so we're not going to be accepting any
                       more requests. */
                    boost::apply_visitor(ql::exc_visitor_t(e, &rg_response->result),
                                         rg.terminal->variant);
                }
            }
        } catch (const runtime_exc_t &e) {
            rg_response->result = e;
        } catch (const ql::exc_t &e) {
            rg_response->result = e;
        } catch (const ql::datum_exc_t &e) {
            rg_response->result = e;
        }
    }

    void operator()(const distribution_read_t &dg) {
        // TODO: do this without copying so much and/or without dynamic memory
        // Sort results by region
        std::vector<distribution_read_response_t> results(count);
        guarantee(count > 0);

        for (size_t i = 0; i < count; ++i) {
            const distribution_read_response_t *result = boost::get<distribution_read_response_t>(&responses[i].response);
            guarantee(result != NULL, "Bad boost::get\n");
            results[i] = *result;
        }

        std::sort(results.begin(), results.end(), distribution_read_response_less_t());

        distribution_read_response_t res;
        size_t i = 0;
        while (i < results.size()) {
            // Find the largest hash shard for this key range
            key_range_t range = results[i].region.inner;
            size_t largest_index = i;
            size_t largest_size = 0;
            size_t total_range_keys = 0;

            while (i < results.size() && results[i].region.inner == range) {
                size_t tmp_total_keys = 0;
                for (std::map<store_key_t, int64_t>::const_iterator mit = results[i].key_counts.begin(); mit != results[i].key_counts.end(); ++mit) {
                    tmp_total_keys += mit->second;
                }

                if (tmp_total_keys > largest_size) {
                    largest_size = tmp_total_keys;
                    largest_index = i;
                }

                total_range_keys += tmp_total_keys;
                ++i;
            }

            if (largest_size > 0) {
                // Scale up the selected hash shard
                double scale_factor = static_cast<double>(total_range_keys) / static_cast<double>(largest_size);

                guarantee(scale_factor >= 1.0);  // Directly provable from code above.

                for (std::map<store_key_t, int64_t>::iterator mit = results[largest_index].key_counts.begin();
                     mit != results[largest_index].key_counts.end();
                     ++mit) {
                    mit->second = static_cast<int64_t>(mit->second * scale_factor);
                }

                res.key_counts.insert(results[largest_index].key_counts.begin(), results[largest_index].key_counts.end());
            }
        }

        // If the result is larger than the requested limit, scale it down
        if (dg.result_limit > 0 && res.key_counts.size() > dg.result_limit) {
            scale_down_distribution(dg.result_limit, &res.key_counts);
        }

        response_out->response = res;
    }

    void operator()(UNUSED const sindex_list_t &sl) {
        guarantee(count == 1);
        guarantee(boost::get<sindex_list_response_t>(&responses[0].response));
        *response_out = responses[0];
    }

private:
    const read_response_t *responses;
    size_t count;
    read_response_t *response_out;
    ql::env_t ql_env;
};

void read_t::unshard(read_response_t *responses, size_t count, read_response_t
        *response, context_t *ctx, signal_t *interruptor) const
    THROWS_ONLY(interrupted_exc_t) {
    rdb_r_unshard_visitor_t v(responses, count, response, ctx, interruptor);
    boost::apply_visitor(v, read);
}

bool rget_data_cmp(const std::pair<store_key_t, boost::shared_ptr<scoped_cJSON_t> >& a,
                   const std::pair<store_key_t, boost::shared_ptr<scoped_cJSON_t> >& b) {
    return a.first < b.first;
}

/* write_t::get_region() implementation */

// TODO: This entire type is suspect, given the performance for batched_replaces_t.  Is it used in
// anything other than assertions?
struct rdb_w_get_region_visitor : public boost::static_visitor<region_t> {
    region_t operator()(const point_replace_t &pr) const {
        return rdb_protocol_t::monokey_region(pr.key);
    }

    static const store_key_t &point_replace_key(const std::pair<int64_t, point_replace_t> &pair) {
        return pair.second.key;
    }

    region_t operator()(const batched_replaces_t &br) const {
        // It shouldn't be empty, but we let the places that would break use a guarantee.
        rassert(!br.point_replaces.empty());
        if (br.point_replaces.empty()) {
            return hash_region_t<key_range_t>();
        }

        store_key_t minimum_key = store_key_t::min();
        store_key_t maximum_key = store_key_t::max();
        uint64_t minimum_hash_value = HASH_REGION_HASH_SIZE - 1;
        uint64_t maximum_hash_value = 0;

        for (auto pair = br.point_replaces.begin(); pair != br.point_replaces.end(); ++pair) {
            if (pair->second.key < minimum_key) {
                minimum_key = pair->second.key;
            }
            if (pair->second.key > maximum_key) {
                maximum_key = pair->second.key;
            }

            const uint64_t hash_value = hash_region_hasher(pair->second.key.contents(), pair->second.key.size());
            if (hash_value < minimum_hash_value) {
                minimum_hash_value = hash_value;
            }
            if (hash_value > maximum_hash_value) {
                maximum_hash_value = hash_value;
            }
        }

        return hash_region_t<key_range_t>(minimum_hash_value, maximum_hash_value + 1,
                                          key_range_t(key_range_t::closed, minimum_key, key_range_t::closed, maximum_key));
    }

    region_t operator()(const point_write_t &pw) const {
        return rdb_protocol_t::monokey_region(pw.key);
    }

    region_t operator()(const point_delete_t &pd) const {
        return rdb_protocol_t::monokey_region(pd.key);
    }

    region_t operator()(const sindex_create_t &s) const {
        return s.region;
    }

    region_t operator()(const sindex_drop_t &d) const {
        return d.region;
    }
};

region_t write_t::get_region() const THROWS_NOTHING {
    return boost::apply_visitor(rdb_w_get_region_visitor(), write);
}

/* write_t::shard implementation */

bool region_contains_key(const hash_region_t<key_range_t> &region, const store_key_t &key) {
    const uint64_t hash_value = hash_region_hasher(key.contents(), key.size());
    if (region.beg <= hash_value && hash_value < region.end) {
        return region.inner.contains_key(key);
    } else {
        return false;
    }
}

struct rdb_w_shard_visitor : public boost::static_visitor<write_t> {
    explicit rdb_w_shard_visitor(const region_t &_region)
        : region(_region)
    { }

    write_t operator()(const point_replace_t &pr) const {
        rassert(rdb_protocol_t::monokey_region(pr.key) == region);
        return write_t(pr);
    }

    write_t operator()(const batched_replaces_t &br) const {
        write_t ret;
        ret.write = batched_replaces_t();
        batched_replaces_t *replaces = boost::get<batched_replaces_t>(&ret.write);

        for (auto point_replace = br.point_replaces.begin(); point_replace != br.point_replaces.end(); ++point_replace) {
            if (region_contains_key(region, point_replace->second.key)) {
                replaces->point_replaces.push_back(*point_replace);
            }
        }

        return ret;
    }

    write_t operator()(const point_write_t &pw) const {
        rassert(rdb_protocol_t::monokey_region(pw.key) == region);
        return write_t(pw);
    }

    write_t operator()(const point_delete_t &pd) const {
        rassert(rdb_protocol_t::monokey_region(pd.key) == region);
        return write_t(pd);
    }

    write_t operator()(const sindex_create_t &c) const {
        sindex_create_t cpy = c;
        cpy.region = region;
        return write_t(cpy);
    }

    write_t operator()(const sindex_drop_t &d) const {
        sindex_drop_t cpy = d;
        cpy.region = region;
        return write_t(cpy);
    }

    const region_t &region;
};


write_t write_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(rdb_w_shard_visitor(region), write);
}

template <class T>
bool first_less(const std::pair<int64_t, T> &left, const std::pair<int64_t, T> &right) {
    return left.first < right.first;
}

struct rdb_w_unshard_visitor_t : public boost::static_visitor<void> {
    void operator()(const point_replace_t &) const { monokey_response(); }

    // The special case here is batched_replaces_response_t, which actually gets sharded into
    // multiple operations instead of getting sent unsplit in a single direction.
    void operator()(const batched_replaces_t &) const {
        std::vector<std::pair<int64_t, point_replace_response_t> > combined;

        for (size_t i = 0; i < count; ++i) {
            const batched_replaces_response_t *batched_response = boost::get<batched_replaces_response_t>(&responses[i].response);
            guarantee(batched_response != NULL, "unsharding nonhomogeneous responses");

            combined.insert(combined.end(),
                            batched_response->point_replace_responses.begin(),
                            batched_response->point_replace_responses.end());
        }

        std::sort(combined.begin(), combined.end(), first_less<point_replace_response_t>);

        *response_out = write_response_t(batched_replaces_response_t(combined));
    }

    void operator()(const point_write_t &) const { monokey_response(); }
    void operator()(const point_delete_t &) const { monokey_response(); }

    void operator()(const sindex_create_t &) const {
        *response_out = responses[0];
    }

    void operator()(const sindex_drop_t &) const {
        *response_out = responses[0];
    }

    rdb_w_unshard_visitor_t(const write_response_t *_responses, size_t _count,
                            write_response_t *_response_out)
        : responses(_responses), count(_count), response_out(_response_out) { }

private:
    void monokey_response() const {
        guarantee(count == 1,
                  "Response with count %zu (greater than 1) returned "
                  "for non-batched write.  (type = %d)",
                  count, responses[0].response.which());

        *response_out = write_response_t(responses[0]);
    }

    const write_response_t *const responses;
    const size_t count;
    write_response_t *const response_out;
};

void write_t::unshard(const write_response_t *responses, size_t count, write_response_t *response_out, context_t *, signal_t *) const THROWS_NOTHING {
    const rdb_w_unshard_visitor_t visitor(responses, count, response_out);
    boost::apply_visitor(visitor, write);
}

store_t::store_t(serializer_t *serializer,
                 const std::string &perfmon_name,
                 int64_t cache_target,
                 bool create,
                 perfmon_collection_t *parent_perfmon_collection,
                 context_t *_ctx,
                 io_backender_t *io,
                 const base_path_t &base_path) :
    btree_store_t<rdb_protocol_t>(serializer, perfmon_name, cache_target,
            create, parent_perfmon_collection, _ctx, io, base_path),
    ctx(_ctx)
{
    // Make sure to continue bringing sindexes up-to-date if it was interrupted earlier

    // This uses a dummy interruptor because this is the only thing using the store at
    //  the moment (since we are still in the constructor), so things should complete
    //  rather quickly.
    cond_t dummy_interruptor;
    read_token_pair_t token_pair;
    new_read_token_pair(&token_pair);

    scoped_ptr_t<transaction_t> txn;
    scoped_ptr_t<real_superblock_t> superblock;
    acquire_superblock_for_read(rwi_read, &token_pair.main_read_token, &txn,
                                &superblock, &dummy_interruptor, false);

    scoped_ptr_t<buf_lock_t> sindex_block;
    acquire_sindex_block_for_read(&token_pair, txn.get(), &sindex_block,
                                  superblock->get_sindex_block_id(),
                                  &dummy_interruptor);

    std::map<std::string, secondary_index_t> sindexes;
    get_secondary_indexes(txn.get(), sindex_block.get(), &sindexes);

    std::set<std::string> sindexes_to_update;
    for (auto it = sindexes.begin(); it != sindexes.end(); ++it) {
        if (!it->second.post_construction_complete) {
            sindexes_to_update.insert(it->first);
        }
    }

    rdb_protocol_details::bring_sindexes_up_to_date(sindexes_to_update, this,
                                                    sindex_block.get(), txn.get());
}

store_t::~store_t() {
    assert_thread();
}

// TODO: get rid of this extra response_t copy on the stack
struct rdb_read_visitor_t : public boost::static_visitor<void> {
    void operator()(const point_read_t &get) {
        response->response = point_read_response_t();
        point_read_response_t *res =
            boost::get<point_read_response_t>(&response->response);
        rdb_get(get.key, btree, txn, superblock, res);
    }

    void operator()(const rget_read_t &rget) {
        if (rget.transform.size() != 0 || rget.terminal) {
            rassert(rget.optargs.size() != 0);
        }
        ql_env.init_optargs(rget.optargs);
        response->response = rget_read_response_t();
        rget_read_response_t *res = boost::get<rget_read_response_t>(&response->response);

        if (!rget.sindex) {
            //Normal rget
            rdb_rget_slice(btree, rget.region.inner, txn, superblock, &ql_env, rget.transform, rget.terminal, res);
        } else {
            scoped_ptr_t<real_superblock_t> sindex_sb;<commit_msg>fixing unit tests by not post constructing sindexes that don't exist<commit_after>// Copyright 2010-2013 RethinkDB, all rights reserved.
#include "rdb_protocol/protocol.hpp"

#include <algorithm>

#include "errors.hpp"
#include <boost/bind.hpp>
#include <boost/make_shared.hpp>

#include "arch/io/disk.hpp"
#include "btree/erase_range.hpp"
#include "btree/parallel_traversal.hpp"
#include "btree/slice.hpp"
#include "btree/superblock.hpp"
#include "clustering/administration/metadata.hpp"
#include "concurrency/cross_thread_watchable.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/vector_stream.hpp"
#include "protob/protob.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/env.hpp"
#include "rdb_protocol/transform_visitors.hpp"
#include "rdb_protocol/pb_utils.hpp"
#include "rdb_protocol/term_walker.hpp"
#include "rpc/semilattice/view/field.hpp"
#include "rpc/semilattice/watchable.hpp"
#include "serializer/config.hpp"

// Ignore clang errors for ql macros
#pragma GCC diagnostic ignored "-Wshadow"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_t;
typedef rdb_protocol_details::range_key_tester_t range_key_tester_t;

typedef rdb_protocol_t::context_t context_t;

typedef rdb_protocol_t::store_t store_t;
typedef rdb_protocol_t::region_t region_t;

typedef rdb_protocol_t::read_t read_t;
typedef rdb_protocol_t::read_response_t read_response_t;

typedef rdb_protocol_t::point_read_t point_read_t;
typedef rdb_protocol_t::point_read_response_t point_read_response_t;

typedef rdb_protocol_t::rget_read_t rget_read_t;
typedef rdb_protocol_t::rget_read_response_t rget_read_response_t;

typedef rdb_protocol_t::distribution_read_t distribution_read_t;
typedef rdb_protocol_t::distribution_read_response_t distribution_read_response_t;

typedef rdb_protocol_t::sindex_list_t sindex_list_t;
typedef rdb_protocol_t::sindex_list_response_t sindex_list_response_t;

typedef rdb_protocol_t::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_t;

typedef rdb_protocol_t::point_replace_t point_replace_t;
typedef rdb_protocol_t::point_replace_response_t point_replace_response_t;

typedef rdb_protocol_t::batched_replaces_t batched_replaces_t;

typedef rdb_protocol_t::point_write_t point_write_t;
typedef rdb_protocol_t::point_write_response_t point_write_response_t;

typedef rdb_protocol_t::point_delete_t point_delete_t;
typedef rdb_protocol_t::point_delete_response_t point_delete_response_t;

typedef rdb_protocol_t::sindex_create_t sindex_create_t;
typedef rdb_protocol_t::sindex_create_response_t sindex_create_response_t;

typedef rdb_protocol_t::sindex_drop_t sindex_drop_t;
typedef rdb_protocol_t::sindex_drop_response_t sindex_drop_response_t;

typedef rdb_protocol_t::backfill_chunk_t backfill_chunk_t;

typedef rdb_protocol_t::backfill_progress_t backfill_progress_t;

typedef rdb_protocol_t::rget_read_response_t::stream_t stream_t;
typedef rdb_protocol_t::rget_read_response_t::groups_t groups_t;
typedef rdb_protocol_t::rget_read_response_t::atom_t atom_t;
typedef rdb_protocol_t::rget_read_response_t::length_t length_t;
typedef rdb_protocol_t::rget_read_response_t::inserted_t inserted_t;

typedef btree_store_t<rdb_protocol_t>::sindex_access_vector_t sindex_access_vector_t;

const std::string rdb_protocol_t::protocol_name("rdb");

RDB_IMPL_PROTOB_SERIALIZABLE(Term);
RDB_IMPL_PROTOB_SERIALIZABLE(Datum);


namespace rdb_protocol_details {

RDB_IMPL_SERIALIZABLE_3(backfill_atom_t, key, value, recency);
RDB_IMPL_SERIALIZABLE_3(transform_atom_t, variant, scopes, backtrace);
RDB_IMPL_SERIALIZABLE_3(terminal_t, variant, scopes, backtrace);

void post_construct_and_drain_queue(
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue,
        auto_drainer_t::lock_t lock)
    THROWS_NOTHING;
/* Creates a queue of operations for the sindex, runs a post construction for
 * the data already in the btree and finally drains the queue. */
void bring_sindexes_up_to_date(
        const std::set<std::string> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        buf_lock_t *sindex_block,
        transaction_t *txn)
    THROWS_NOTHING
{
    /* We register our modification queue here. An important point about
     * correctness here: we've held the superblock this whole time and will
     * continue to do so until the call to post_construct_secondary_indexes
     * begins a parallel traversal which releases the superblock. This
     * serves to make sure that every changes which we don't learn about in
     * the parallel traversal we do learn about from the mod queue. */
    uuid_u post_construct_id = generate_uuid();

    boost::shared_ptr<internal_disk_backed_queue_t> mod_queue(
            new internal_disk_backed_queue_t(
                store->io_backender_,
                serializer_filepath_t(store->base_path_, "post_construction_" + uuid_to_str(post_construct_id)),
                &store->perfmon_collection));

    {
        mutex_t::acq_t acq;
        store->lock_sindex_queue(sindex_block, &acq);
        store->register_sindex_queue(mod_queue.get(), &acq);
    }

    std::map<std::string, secondary_index_t> sindexes;
    store->get_sindexes(sindex_block, txn, &sindexes);
    std::set<uuid_u> sindexes_to_bring_up_to_date_uuid;

    for (auto it = sindexes_to_bring_up_to_date.begin(); 
         it != sindexes_to_bring_up_to_date.end(); ++it) {
        guarantee(std_contains(sindexes, *it));
        sindexes_to_bring_up_to_date_uuid.insert(sindexes[*it].id);
    }

    coro_t::spawn_sometime(boost::bind(
                &post_construct_and_drain_queue,
                sindexes_to_bring_up_to_date_uuid,
                store,
                mod_queue,
                auto_drainer_t::lock_t(&store->drainer)));
}

/* This function is really part of the logic of bring_sindexes_up_to_date
 * however it needs to be in a seperate function so that it can be spawned in a
 * coro. */
void post_construct_and_drain_queue(
        const std::set<uuid_u> &sindexes_to_bring_up_to_date,
        btree_store_t<rdb_protocol_t> *store,
        boost::shared_ptr<internal_disk_backed_queue_t> mod_queue,
        auto_drainer_t::lock_t lock)
    THROWS_NOTHING
{
    try {
        post_construct_secondary_indexes(store, sindexes_to_bring_up_to_date, lock.get_drain_signal());

        /* Drain the queue. */

        int previous_size = mod_queue->size();
        while (!lock.get_drain_signal()->is_pulsed()) {
            write_token_pair_t token_pair;
            store->new_write_token_pair(&token_pair);

            scoped_ptr_t<transaction_t> queue_txn;
            scoped_ptr_t<real_superblock_t> queue_superblock;

            // We don't need hard durability here, because a secondary index just gets rebuilt
            // if the server dies while it's partially constructed.
            store->acquire_superblock_for_write(
                rwi_write,
                repli_timestamp_t::distant_past,
                2,
                WRITE_DURABILITY_SOFT,
                &token_pair.main_write_token,
                &queue_txn,
                &queue_superblock,
                lock.get_drain_signal());

            scoped_ptr_t<buf_lock_t> queue_sindex_block;
            store->acquire_sindex_block_for_write(
                &token_pair,
                queue_txn.get(),
                &queue_sindex_block,
                queue_superblock->get_sindex_block_id(),
                lock.get_drain_signal());

            sindex_access_vector_t sindexes;
            store->acquire_sindex_superblocks_for_write(
                    sindexes_to_bring_up_to_date,
                    queue_sindex_block.get(),
                    queue_txn.get(),
                    &sindexes);

            if (sindexes.empty()) {
                break;
            }

            mutex_t::acq_t acq;
            store->lock_sindex_queue(queue_sindex_block.get(), &acq);

            while (mod_queue->size() >= previous_size &&
                   mod_queue->size() > 0) {
                std::vector<char> data_vec;
                mod_queue->pop(&data_vec);
                vector_read_stream_t read_stream(&data_vec);

                rdb_modification_report_t mod_report;
                int ser_res = deserialize(&read_stream, &mod_report);
                guarantee_err(ser_res == 0, "corruption in disk-backed queue");

                rdb_update_sindexes(sindexes, &mod_report, queue_txn.get());
            }

            previous_size = mod_queue->size();

            if (mod_queue->size() == 0) {
                for (auto it = sindexes_to_bring_up_to_date.begin();
                     it != sindexes_to_bring_up_to_date.end(); ++it) {
                    store->mark_index_up_to_date(*it, queue_txn.get(), queue_sindex_block.get());
                }
                store->deregister_sindex_queue(mod_queue.get(), &acq);
                return;
            }
        }
    } catch (const interrupted_exc_t &) {
        // We were interrupted so we just exit. Sindex post construct is in an
        // indeterminate state and will be cleaned up at a later point.
    }

    {
        /* If we get here it's either because we were interrupted or the
         * sindexes we were post constructiing were deleted. Either way we need
         * to clean up the queue. */
        write_token_pair_t token_pair;
        store->new_write_token_pair(&token_pair);

        scoped_ptr_t<transaction_t> queue_txn;
        scoped_ptr_t<real_superblock_t> queue_superblock;

        store->acquire_superblock_for_write(
            rwi_write,
            repli_timestamp_t::distant_past,
            2,
            WRITE_DURABILITY_HARD,
            &token_pair.main_write_token,
            &queue_txn,
            &queue_superblock,
            lock.get_drain_signal());

        scoped_ptr_t<buf_lock_t> queue_sindex_block;
        store->acquire_sindex_block_for_write(
            &token_pair,
            queue_txn.get(),
            &queue_sindex_block,
            queue_superblock->get_sindex_block_id(),
            lock.get_drain_signal());

        mutex_t::acq_t acq;
        store->lock_sindex_queue(queue_sindex_block.get(), &acq);
        store->deregister_sindex_queue(mod_queue.get(), &acq);
    }
}


bool range_key_tester_t::key_should_be_erased(const btree_key_t *key) {
    uint64_t h = hash_region_hasher(key->contents, key->size);
    return delete_range->beg <= h && h < delete_range->end
        && delete_range->inner.contains_key(key->contents, key->size);
}

}  // namespace rdb_protocol_details

rdb_protocol_t::context_t::context_t()
    : pool_group(NULL), ns_repo(NULL),
    cross_thread_namespace_watchables(get_num_threads()),
    cross_thread_database_watchables(get_num_threads()),
    directory_read_manager(NULL),
    signals(get_num_threads())
{ }

rdb_protocol_t::context_t::context_t(
    extproc::pool_group_t *_pool_group,
    namespace_repo_t<rdb_protocol_t> *_ns_repo,
    boost::shared_ptr<semilattice_readwrite_view_t<cluster_semilattice_metadata_t> >
        _semilattice_metadata,
    directory_read_manager_t<cluster_directory_metadata_t>
        *_directory_read_manager,
    machine_id_t _machine_id)
    : pool_group(_pool_group), ns_repo(_ns_repo),
      cross_thread_namespace_watchables(get_num_threads()),
      cross_thread_database_watchables(get_num_threads()),
      semilattice_metadata(_semilattice_metadata),
      directory_read_manager(_directory_read_manager),
      signals(get_num_threads()),
      machine_id(_machine_id)
{
    for (int thread = 0; thread < get_num_threads(); ++thread) {
        cross_thread_namespace_watchables[thread].init(new cross_thread_watchable_variable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > >(
                                                    clone_ptr_t<semilattice_watchable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > > >
                                                        (new semilattice_watchable_t<cow_ptr_t<namespaces_semilattice_metadata_t<rdb_protocol_t> > >(
                                                            metadata_field(&cluster_semilattice_metadata_t::rdb_namespaces, _semilattice_metadata))), thread));

        cross_thread_database_watchables[thread].init(new cross_thread_watchable_variable_t<databases_semilattice_metadata_t>(
                                                    clone_ptr_t<semilattice_watchable_t<databases_semilattice_metadata_t> >
                                                        (new semilattice_watchable_t<databases_semilattice_metadata_t>(
                                                            metadata_field(&cluster_semilattice_metadata_t::databases, _semilattice_metadata))), thread));

        signals[thread].init(new cross_thread_signal_t(&interruptor, thread));
    }
}

rdb_protocol_t::context_t::~context_t() { }

// Construct a region containing only the specified key
region_t rdb_protocol_t::monokey_region(const store_key_t &k) {
    uint64_t h = hash_region_hasher(k.contents(), k.size());
    return region_t(h, h + 1, key_range_t(key_range_t::closed, k, key_range_t::closed, k));
}

key_range_t rdb_protocol_t::sindex_key_range(const store_key_t &start, const store_key_t &end) {
    std::string start_key_str(key_to_unescaped_str(start));
    std::string end_key_str(key_to_unescaped_str(end));
    store_key_t end_key;

    // Need to make the next largest store_key_t without making the key longer
    while (end_key_str.length() > 0 &&
           end_key_str[end_key_str.length() - 1] == static_cast<char>(255)) {
        end_key_str.erase(end_key_str.length() - 1);
    }

    if (end_key_str.length() == 0) {
        end_key = store_key_t::max();
    } else {
        ++end_key_str[end_key_str.length() - 1];
        end_key = store_key_t(end_key_str);
    }

    return key_range_t(key_range_t::closed, store_key_t(start_key_str),
                       key_range_t::open, end_key);
}

/* read_t::get_region implementation */
struct rdb_r_get_region_visitor : public boost::static_visitor<region_t> {
    region_t operator()(const point_read_t &pr) const {
        return rdb_protocol_t::monokey_region(pr.key);
    }

    region_t operator()(const rget_read_t &rg) const {
        return rg.region;
    }

    region_t operator()(const distribution_read_t &dg) const {
        return dg.region;
    }

    region_t operator()(UNUSED const sindex_list_t &sl) const {
        return rdb_protocol_t::monokey_region(store_key_t());
    }
};

region_t read_t::get_region() const THROWS_NOTHING {
    return boost::apply_visitor(rdb_r_get_region_visitor(), read);
}

struct rdb_r_shard_visitor : public boost::static_visitor<read_t> {
    explicit rdb_r_shard_visitor(const region_t &_region)
        : region(_region)
    { }

    read_t operator()(const point_read_t &pr) const {
        rassert(rdb_protocol_t::monokey_region(pr.key) == region);
        return read_t(pr);
    }

    read_t operator()(const rget_read_t &rg) const {
        rassert(region_is_superset(rg.region, region));
        rget_read_t _rg(rg);
        _rg.region = region;
        return read_t(_rg);
    }

    read_t operator()(const distribution_read_t &dg) const {
        rassert(region_is_superset(dg.region, region));
        distribution_read_t _dg(dg);
        _dg.region = region;
        return read_t(_dg);
    }

    read_t operator()(const sindex_list_t &sl) const {
        return read_t(sl);
    }

    const region_t &region;
};

read_t read_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(rdb_r_shard_visitor(region), read);
}

/* read_t::unshard implementation */
bool read_response_cmp(const read_response_t &l, const read_response_t &r) {
    const rget_read_response_t *lr = boost::get<rget_read_response_t>(&l.response);
    guarantee(lr);
    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&r.response);
    guarantee(rr);
    return lr->key_range < rr->key_range;
}

/* A visitor to handle this unsharding process for us. */

class distribution_read_response_less_t {
public:
    bool operator()(const distribution_read_response_t& x, const distribution_read_response_t& y) {
        if (x.region.inner == y.region.inner) {
            return x.region < y.region;
        } else {
            return x.region.inner < y.region.inner;
        }
    }
};

// Scale the distribution down by combining ranges to fit it within the limit of
// the query
void scale_down_distribution(size_t result_limit, std::map<store_key_t, int64_t> *key_counts) {
    guarantee(result_limit > 0);
    const size_t combine = (key_counts->size() / result_limit); // Combine this many other ranges into the previous range
    for (std::map<store_key_t, int64_t>::iterator it = key_counts->begin(); it != key_counts->end(); ) {
        std::map<store_key_t, int64_t>::iterator next = it;
        ++next;
        for (size_t i = 0; i < combine && next != key_counts->end(); ++i) {
            it->second += next->second;
            std::map<store_key_t, int64_t>::iterator tmp = next;
            ++next;
            key_counts->erase(tmp);
        }
        it = next;
    }
}

class rdb_r_unshard_visitor_t : public boost::static_visitor<void> {
public:
    rdb_r_unshard_visitor_t(const read_response_t *_responses,
                            size_t _count,
                            read_response_t *_response_out,
                            rdb_protocol_t::context_t *ctx,
                            signal_t *interruptor)
        : responses(_responses), count(_count), response_out(_response_out),
          ql_env(ctx->pool_group,
                 ctx->ns_repo,
                 ctx->cross_thread_namespace_watchables[get_thread_id()].get()
                     ->get_watchable(),
                 ctx->cross_thread_database_watchables[get_thread_id()].get()
                     ->get_watchable(),
                 ctx->semilattice_metadata,
                 NULL,
                 boost::make_shared<js::runner_t>(),
                 interruptor,
                 ctx->machine_id,
                 std::map<std::string, ql::wire_func_t>())
    { }

    void operator()(const point_read_t &) {
        guarantee(count == 1);
        guarantee(NULL != boost::get<point_read_response_t>(&responses[0].response));
        *response_out = responses[0];
    }

    void operator()(const rget_read_t &rg) {
        response_out->response = rget_read_response_t();
        rget_read_response_t *rg_response = boost::get<rget_read_response_t>(&response_out->response);
        rg_response->truncated = false;
        rg_response->key_range = read_t(rg).get_region().inner;
        rg_response->last_considered_key = read_t(rg).get_region().inner.left;

        try {
            /* First check to see if any of the responses we're unsharding threw. */
            for (size_t i = 0; i < count; ++i) {
                // TODO: we're ignoring the limit when recombining.
                const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                guarantee(rr != NULL);

                if (const runtime_exc_t *e = boost::get<runtime_exc_t>(&rr->result)) {
                    throw *e;
                } else if (const ql::exc_t *e2 = boost::get<ql::exc_t>(&rr->result)) {
                    throw *e2;
                } else if (const ql::datum_exc_t *e3 = boost::get<ql::datum_exc_t>(&rr->result)) {
                    throw *e3;
                }
            }

            if (!rg.terminal) {
                //A vanilla range get
                //First we need to determine the cutoff key:
                rg_response->last_considered_key = store_key_t::max();
                for (size_t i = 0; i < count; ++i) {
                    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                    guarantee(rr != NULL);

                    if (rg_response->last_considered_key > rr->last_considered_key && rr->truncated) {
                        rg_response->last_considered_key = rr->last_considered_key;
                    }
                }

                rg_response->result = stream_t();
                stream_t *res_stream = boost::get<stream_t>(&rg_response->result);
                for (size_t i = 0; i < count; ++i) {
                    // TODO: we're ignoring the limit when recombining.
                    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&responses[i].response);
                    guarantee(rr != NULL);

                    const stream_t *stream = boost::get<stream_t>(&(rr->result));

                    for (stream_t::const_iterator it = stream->begin(); it != stream->end(); ++it) {
                        if (it->first <= rg_response->last_considered_key) {
                            res_stream->push_back(*it);
                        }
                    }

                    rg_response->truncated = rg_response->truncated || rr->truncated;
                }
            } else {
                try {
                    if (const ql::reduce_wire_func_t *reduce_func =
                            boost::get<ql::reduce_wire_func_t>(&rg.terminal->variant)) {
                        ql::reduce_wire_func_t local_reduce_func = *reduce_func;
                        rg_response->result = rget_read_response_t::empty_t();
                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            ql::wire_datum_t *lhs = boost::get<ql::wire_datum_t>(&rg_response->result);
                            const ql::wire_datum_t *rhs = boost::get<ql::wire_datum_t>(&(_rr->result));
                            if (!rhs) {
                                guarantee(boost::get<rget_read_response_t::empty_t>(&(_rr->result)));
                                continue;
                            } else {
                                ql::wire_datum_t local_rhs = *rhs;
                                if (lhs) {
                                    const ql::datum_t *reduced_val =
                                        local_reduce_func.
                                            compile(&ql_env)->
                                                call(lhs->compile(&ql_env), local_rhs.compile(&ql_env))->
                                                    as_datum();
                                    rg_response->result = ql::wire_datum_t(reduced_val);
                                } else {
                                    guarantee(boost::get<rget_read_response_t::empty_t>(&rg_response->result));
                                    rg_response->result = _rr->result;
                                }
                            }
                        }
                        ql::wire_datum_t *final_val = boost::get<ql::wire_datum_t>(&rg_response->result);
                        if (final_val) {
                            final_val->finalize();
                        }
                    } else if (boost::get<ql::count_wire_func_t>(&rg.terminal->variant)) {
                        rg_response->result =
                            ql::wire_datum_t(ql_env.add_ptr(new ql::datum_t(0.0)));

                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            ql::wire_datum_t *lhs = boost::get<ql::wire_datum_t>(&rg_response->result);
                            const ql::wire_datum_t *rhs = boost::get<ql::wire_datum_t>(&(_rr->result));
                            ql::wire_datum_t local_rhs = *rhs;

                            const ql::datum_t *sum =
                                ql_env.add_ptr(new ql::datum_t(
                                                   lhs->compile(&ql_env)->as_num() +
                                                   local_rhs.compile(&ql_env)->as_num()));
                            rg_response->result = ql::wire_datum_t(sum);
                        }
                        boost::get<ql::wire_datum_t>(rg_response->result).finalize();
                    } else if (const ql::gmr_wire_func_t *gmr_func =
                            boost::get<ql::gmr_wire_func_t>(&rg.terminal->variant)) {
                        ql::gmr_wire_func_t local_gmr_func = *gmr_func;
                        rg_response->result = ql::wire_datum_map_t();
                        ql::wire_datum_map_t *map =
                            boost::get<ql::wire_datum_map_t>(&rg_response->result);

                        for (size_t i = 0; i < count; ++i) {
                            const rget_read_response_t *_rr =
                                boost::get<rget_read_response_t>(&responses[i].response);
                            guarantee(_rr);
                            const ql::wire_datum_map_t *rhs =
                                boost::get<ql::wire_datum_map_t>(&(_rr->result));
                            r_sanity_check(rhs);
                            ql::wire_datum_map_t local_rhs = *rhs;
                            local_rhs.compile(&ql_env);

                            const ql::datum_t *rhs_arr = local_rhs.to_arr(&ql_env);
                            for (size_t f = 0; f < rhs_arr->size(); ++f) {
                                const ql::datum_t *key = rhs_arr->get(f)->get("group");
                                const ql::datum_t *val = rhs_arr->get(f)->get("reduction");
                                if (!map->has(key)) {
                                    map->set(key, val);
                                } else {
                                    ql::func_t *r = local_gmr_func.compile_reduce(&ql_env);
                                    map->set(key, r->call(map->get(key), val)->as_datum());
                                }
                            }
                        }
                        boost::get<ql::wire_datum_map_t>(rg_response->result).finalize();
                    } else {
                        unreachable();
                    }
                } catch (const ql::datum_exc_t &e) {
                    /* Evaluation threw so we're not going to be accepting any
                       more requests. */
                    boost::apply_visitor(ql::exc_visitor_t(e, &rg_response->result),
                                         rg.terminal->variant);
                }
            }
        } catch (const runtime_exc_t &e) {
            rg_response->result = e;
        } catch (const ql::exc_t &e) {
            rg_response->result = e;
        } catch (const ql::datum_exc_t &e) {
            rg_response->result = e;
        }
    }

    void operator()(const distribution_read_t &dg) {
        // TODO: do this without copying so much and/or without dynamic memory
        // Sort results by region
        std::vector<distribution_read_response_t> results(count);
        guarantee(count > 0);

        for (size_t i = 0; i < count; ++i) {
            const distribution_read_response_t *result = boost::get<distribution_read_response_t>(&responses[i].response);
            guarantee(result != NULL, "Bad boost::get\n");
            results[i] = *result;
        }

        std::sort(results.begin(), results.end(), distribution_read_response_less_t());

        distribution_read_response_t res;
        size_t i = 0;
        while (i < results.size()) {
            // Find the largest hash shard for this key range
            key_range_t range = results[i].region.inner;
            size_t largest_index = i;
            size_t largest_size = 0;
            size_t total_range_keys = 0;

            while (i < results.size() && results[i].region.inner == range) {
                size_t tmp_total_keys = 0;
                for (std::map<store_key_t, int64_t>::const_iterator mit = results[i].key_counts.begin(); mit != results[i].key_counts.end(); ++mit) {
                    tmp_total_keys += mit->second;
                }

                if (tmp_total_keys > largest_size) {
                    largest_size = tmp_total_keys;
                    largest_index = i;
                }

                total_range_keys += tmp_total_keys;
                ++i;
            }

            if (largest_size > 0) {
                // Scale up the selected hash shard
                double scale_factor = static_cast<double>(total_range_keys) / static_cast<double>(largest_size);

                guarantee(scale_factor >= 1.0);  // Directly provable from code above.

                for (std::map<store_key_t, int64_t>::iterator mit = results[largest_index].key_counts.begin();
                     mit != results[largest_index].key_counts.end();
                     ++mit) {
                    mit->second = static_cast<int64_t>(mit->second * scale_factor);
                }

                res.key_counts.insert(results[largest_index].key_counts.begin(), results[largest_index].key_counts.end());
            }
        }

        // If the result is larger than the requested limit, scale it down
        if (dg.result_limit > 0 && res.key_counts.size() > dg.result_limit) {
            scale_down_distribution(dg.result_limit, &res.key_counts);
        }

        response_out->response = res;
    }

    void operator()(UNUSED const sindex_list_t &sl) {
        guarantee(count == 1);
        guarantee(boost::get<sindex_list_response_t>(&responses[0].response));
        *response_out = responses[0];
    }

private:
    const read_response_t *responses;
    size_t count;
    read_response_t *response_out;
    ql::env_t ql_env;
};

void read_t::unshard(read_response_t *responses, size_t count, read_response_t
        *response, context_t *ctx, signal_t *interruptor) const
    THROWS_ONLY(interrupted_exc_t) {
    rdb_r_unshard_visitor_t v(responses, count, response, ctx, interruptor);
    boost::apply_visitor(v, read);
}

bool rget_data_cmp(const std::pair<store_key_t, boost::shared_ptr<scoped_cJSON_t> >& a,
                   const std::pair<store_key_t, boost::shared_ptr<scoped_cJSON_t> >& b) {
    return a.first < b.first;
}

/* write_t::get_region() implementation */

// TODO: This entire type is suspect, given the performance for batched_replaces_t.  Is it used in
// anything other than assertions?
struct rdb_w_get_region_visitor : public boost::static_visitor<region_t> {
    region_t operator()(const point_replace_t &pr) const {
        return rdb_protocol_t::monokey_region(pr.key);
    }

    static const store_key_t &point_replace_key(const std::pair<int64_t, point_replace_t> &pair) {
        return pair.second.key;
    }

    region_t operator()(const batched_replaces_t &br) const {
        // It shouldn't be empty, but we let the places that would break use a guarantee.
        rassert(!br.point_replaces.empty());
        if (br.point_replaces.empty()) {
            return hash_region_t<key_range_t>();
        }

        store_key_t minimum_key = store_key_t::min();
        store_key_t maximum_key = store_key_t::max();
        uint64_t minimum_hash_value = HASH_REGION_HASH_SIZE - 1;
        uint64_t maximum_hash_value = 0;

        for (auto pair = br.point_replaces.begin(); pair != br.point_replaces.end(); ++pair) {
            if (pair->second.key < minimum_key) {
                minimum_key = pair->second.key;
            }
            if (pair->second.key > maximum_key) {
                maximum_key = pair->second.key;
            }

            const uint64_t hash_value = hash_region_hasher(pair->second.key.contents(), pair->second.key.size());
            if (hash_value < minimum_hash_value) {
                minimum_hash_value = hash_value;
            }
            if (hash_value > maximum_hash_value) {
                maximum_hash_value = hash_value;
            }
        }

        return hash_region_t<key_range_t>(minimum_hash_value, maximum_hash_value + 1,
                                          key_range_t(key_range_t::closed, minimum_key, key_range_t::closed, maximum_key));
    }

    region_t operator()(const point_write_t &pw) const {
        return rdb_protocol_t::monokey_region(pw.key);
    }

    region_t operator()(const point_delete_t &pd) const {
        return rdb_protocol_t::monokey_region(pd.key);
    }

    region_t operator()(const sindex_create_t &s) const {
        return s.region;
    }

    region_t operator()(const sindex_drop_t &d) const {
        return d.region;
    }
};

region_t write_t::get_region() const THROWS_NOTHING {
    return boost::apply_visitor(rdb_w_get_region_visitor(), write);
}

/* write_t::shard implementation */

bool region_contains_key(const hash_region_t<key_range_t> &region, const store_key_t &key) {
    const uint64_t hash_value = hash_region_hasher(key.contents(), key.size());
    if (region.beg <= hash_value && hash_value < region.end) {
        return region.inner.contains_key(key);
    } else {
        return false;
    }
}

struct rdb_w_shard_visitor : public boost::static_visitor<write_t> {
    explicit rdb_w_shard_visitor(const region_t &_region)
        : region(_region)
    { }

    write_t operator()(const point_replace_t &pr) const {
        rassert(rdb_protocol_t::monokey_region(pr.key) == region);
        return write_t(pr);
    }

    write_t operator()(const batched_replaces_t &br) const {
        write_t ret;
        ret.write = batched_replaces_t();
        batched_replaces_t *replaces = boost::get<batched_replaces_t>(&ret.write);

        for (auto point_replace = br.point_replaces.begin(); point_replace != br.point_replaces.end(); ++point_replace) {
            if (region_contains_key(region, point_replace->second.key)) {
                replaces->point_replaces.push_back(*point_replace);
            }
        }

        return ret;
    }

    write_t operator()(const point_write_t &pw) const {
        rassert(rdb_protocol_t::monokey_region(pw.key) == region);
        return write_t(pw);
    }

    write_t operator()(const point_delete_t &pd) const {
        rassert(rdb_protocol_t::monokey_region(pd.key) == region);
        return write_t(pd);
    }

    write_t operator()(const sindex_create_t &c) const {
        sindex_create_t cpy = c;
        cpy.region = region;
        return write_t(cpy);
    }

    write_t operator()(const sindex_drop_t &d) const {
        sindex_drop_t cpy = d;
        cpy.region = region;
        return write_t(cpy);
    }

    const region_t &region;
};


write_t write_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(rdb_w_shard_visitor(region), write);
}

template <class T>
bool first_less(const std::pair<int64_t, T> &left, const std::pair<int64_t, T> &right) {
    return left.first < right.first;
}

struct rdb_w_unshard_visitor_t : public boost::static_visitor<void> {
    void operator()(const point_replace_t &) const { monokey_response(); }

    // The special case here is batched_replaces_response_t, which actually gets sharded into
    // multiple operations instead of getting sent unsplit in a single direction.
    void operator()(const batched_replaces_t &) const {
        std::vector<std::pair<int64_t, point_replace_response_t> > combined;

        for (size_t i = 0; i < count; ++i) {
            const batched_replaces_response_t *batched_response = boost::get<batched_replaces_response_t>(&responses[i].response);
            guarantee(batched_response != NULL, "unsharding nonhomogeneous responses");

            combined.insert(combined.end(),
                            batched_response->point_replace_responses.begin(),
                            batched_response->point_replace_responses.end());
        }

        std::sort(combined.begin(), combined.end(), first_less<point_replace_response_t>);

        *response_out = write_response_t(batched_replaces_response_t(combined));
    }

    void operator()(const point_write_t &) const { monokey_response(); }
    void operator()(const point_delete_t &) const { monokey_response(); }

    void operator()(const sindex_create_t &) const {
        *response_out = responses[0];
    }

    void operator()(const sindex_drop_t &) const {
        *response_out = responses[0];
    }

    rdb_w_unshard_visitor_t(const write_response_t *_responses, size_t _count,
                            write_response_t *_response_out)
        : responses(_responses), count(_count), response_out(_response_out) { }

private:
    void monokey_response() const {
        guarantee(count == 1,
                  "Response with count %zu (greater than 1) returned "
                  "for non-batched write.  (type = %d)",
                  count, responses[0].response.which());

        *response_out = write_response_t(responses[0]);
    }

    const write_response_t *const responses;
    const size_t count;
    write_response_t *const response_out;
};

void write_t::unshard(const write_response_t *responses, size_t count, write_response_t *response_out, context_t *, signal_t *) const THROWS_NOTHING {
    const rdb_w_unshard_visitor_t visitor(responses, count, response_out);
    boost::apply_visitor(visitor, write);
}

store_t::store_t(serializer_t *serializer,
                 const std::string &perfmon_name,
                 int64_t cache_target,
                 bool create,
                 perfmon_collection_t *parent_perfmon_collection,
                 context_t *_ctx,
                 io_backender_t *io,
                 const base_path_t &base_path) :
    btree_store_t<rdb_protocol_t>(serializer, perfmon_name, cache_target,
            create, parent_perfmon_collection, _ctx, io, base_path),
    ctx(_ctx)
{
    // Make sure to continue bringing sindexes up-to-date if it was interrupted earlier

    // This uses a dummy interruptor because this is the only thing using the store at
    //  the moment (since we are still in the constructor), so things should complete
    //  rather quickly.
    cond_t dummy_interruptor;
    read_token_pair_t token_pair;
    new_read_token_pair(&token_pair);

    scoped_ptr_t<transaction_t> txn;
    scoped_ptr_t<real_superblock_t> superblock;
    acquire_superblock_for_read(rwi_read, &token_pair.main_read_token, &txn,
                                &superblock, &dummy_interruptor, false);

    scoped_ptr_t<buf_lock_t> sindex_block;
    acquire_sindex_block_for_read(&token_pair, txn.get(), &sindex_block,
                                  superblock->get_sindex_block_id(),
                                  &dummy_interruptor);

    std::map<std::string, secondary_index_t> sindexes;
    get_secondary_indexes(txn.get(), sindex_block.get(), &sindexes);

    std::set<std::string> sindexes_to_update;
    for (auto it = sindexes.begin(); it != sindexes.end(); ++it) {
        if (!it->second.post_construction_complete) {
            sindexes_to_update.insert(it->first);
        }
    }

    if (!sindexes_to_update.empty()) {
        rdb_protocol_details::bring_sindexes_up_to_date(sindexes_to_update, this,
                                                        sindex_block.get(), txn.get());
    }
}

store_t::~store_t() {
    assert_thread();
}

// TODO: get rid of this extra response_t copy on the stack
struct rdb_read_visitor_t : public boost::static_visitor<void> {
    void operator()(const point_read_t &get) {
        response->response = point_read_response_t();
        point_read_response_t *res =
            boost::get<point_read_response_t>(&response->response);
        rdb_get(get.key, btree, txn, superblock, res);
    }

    void operator()(const rget_read_t &rget) {
        if (rget.transform.size() != 0 || rget.terminal) {
            rassert(rget.optargs.size() != 0);
        }
        ql_env.init_optargs(rget.optargs);
        response->response = rget_read_response_t();
        rget_read_response_t *res = boost::get<rget_read_response_t>(&response->response);

        if (!rget.sindex) {
            //Normal rget
            rdb_rget_slice(btree, rget.region.inner, txn, superblock, &ql_env, rget.transform, rget.terminal, res);
        } else {
            scoped_ptr_t<real_superblock_t> sindex_sb;
{'fraction_non_alphanumeric': 0.10204711576451996, 'fraction_numerical': 0.002592848676412485, 'mean_word_length': 2.3862781168994065, 'pattern_counts': {'":': 0, '<': 285, '<?xml version=': 0, '>': 545, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 84, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3282283', 'n_tokens_mistral': 26157, 'n_tokens_neox': 25181, 'n_words': 4339}