<commit_before>#include "errors.hpp"
#include <boost/bind.hpp>
#include <boost/function.hpp>

#include "btree/erase_range.hpp"
#include "btree/slice.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/vector_stream.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/protocol.hpp"
#include "rdb_protocol/query_language.hpp"
#include "serializer/config.hpp"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_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::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_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::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;

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

// 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));
}

/* read_t::get_region implementation */
struct 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 {
        // TODO: Sam bets this causes problems
        return region_t(rg.key_range);
    }
};

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

/* read_t::shard implementation */

struct r_shard_visitor : public boost::static_visitor<read_t> {
    explicit 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(region_t(rg.key_range), region));
        // TODO: Reevaluate this code.  Should rget_query_t really have a region_t range?
        rget_read_t _rg(rg);
        _rg.key_range = region.inner;
        return read_t(_rg);
    }
    const region_t &region;
};

read_t read_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(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);
    rassert(lr);
    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&r.response);
    rassert(rr);
    return lr->key_range < rr->key_range;
}

read_response_t read_t::unshard(std::vector<read_response_t> responses, temporary_cache_t *) const THROWS_NOTHING {
    const point_read_t *pr = boost::get<point_read_t>(&read);
    if (pr) {
        rassert(responses.size() == 1);
        rassert(boost::get<point_read_response_t>(&responses[0].response));
        return responses[0];
    }

    const rget_read_t *rg = boost::get<rget_read_t>(&read);
    if (rg) {
        std::sort(responses.begin(), responses.end(), read_response_cmp);
        rget_read_response_t rg_response;
        rg_response.truncated = false;
        rg_response.key_range = get_region().inner;
        typedef std::vector<read_response_t>::iterator rri_t;

        if (!rg->terminal) {
            rg_response.result = stream_t();
            stream_t *res_stream = boost::get<stream_t>(&rg_response.result);
            for(rri_t i = responses.begin(); i != responses.end(); ++i) {
                // TODO: we're ignoring the limit when recombining.
                const rget_read_response_t *_rr = boost::get<rget_read_response_t>(&i->response);
                rassert(_rr);

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

                res_stream->insert(res_stream->end(), stream->begin(), stream->end());
                rg_response.truncated = rg_response.truncated || _rr->truncated;
            }
        } else if (const Builtin_GroupedMapReduce *gmr = boost::get<Builtin_GroupedMapReduce>(&*rg->terminal)) {
            //Mapreduce
            rg_response.result = groups_t();
            groups_t *res_groups = boost::get<groups_t>(&rg_response.result);
            for(rri_t i = responses.begin(); i != responses.end(); ++i) {
                const rget_read_response_t *_rr = boost::get<rget_read_response_t>(&i->response);
                guarantee(_rr);

                const groups_t *groups = boost::get<groups_t>(&(_rr->result));
                query_language::runtime_environment_t *env = NULL; //obviously this is a problem
                query_language::backtrace_t backtrace;

                for (groups_t::const_iterator j = groups->begin(); j != groups->end(); ++j) {
                    query_language::new_val_scope_t scope(&env->scope);
                    Term base = gmr->reduction().base(),
                         body = gmr->reduction().body();

                    env->scope.put_in_scope(gmr->reduction().var1(), get_with_default(*res_groups, j->first, eval(&base, env, backtrace)));
                    env->scope.put_in_scope(gmr->reduction().var2(), j->second);

                    (*res_groups)[j->first] = eval(&body, env, backtrace);
                }
            }
        } else if (boost::get<Reduction>(&*rg->terminal)) {
        } else if (boost::get<rdb_protocol_details::Length>(&*rg->terminal)) {
        } else if (boost::get<WriteQuery_ForEach>(&*rg->terminal)) {
        }
        return read_response_t(rg_response);
    }

    unreachable("Unknown read response.");
}

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;<commit_msg>Adds unsharding of reduce queries.<commit_after>#include "errors.hpp"
#include <boost/bind.hpp>
#include <boost/function.hpp>

#include "btree/erase_range.hpp"
#include "btree/slice.hpp"
#include "concurrency/pmap.hpp"
#include "concurrency/wait_any.hpp"
#include "containers/archive/vector_stream.hpp"
#include "rdb_protocol/btree.hpp"
#include "rdb_protocol/protocol.hpp"
#include "rdb_protocol/query_language.hpp"
#include "serializer/config.hpp"

typedef rdb_protocol_details::backfill_atom_t rdb_backfill_atom_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::write_t write_t;
typedef rdb_protocol_t::write_response_t write_response_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::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;

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

// 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));
}

/* read_t::get_region implementation */
struct 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 {
        // TODO: Sam bets this causes problems
        return region_t(rg.key_range);
    }
};

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

/* read_t::shard implementation */

struct r_shard_visitor : public boost::static_visitor<read_t> {
    explicit 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(region_t(rg.key_range), region));
        // TODO: Reevaluate this code.  Should rget_query_t really have a region_t range?
        rget_read_t _rg(rg);
        _rg.key_range = region.inner;
        return read_t(_rg);
    }
    const region_t &region;
};

read_t read_t::shard(const region_t &region) const THROWS_NOTHING {
    return boost::apply_visitor(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);
    rassert(lr);
    const rget_read_response_t *rr = boost::get<rget_read_response_t>(&r.response);
    rassert(rr);
    return lr->key_range < rr->key_range;
}

read_response_t read_t::unshard(std::vector<read_response_t> responses, temporary_cache_t *) const THROWS_NOTHING {
    const point_read_t *pr = boost::get<point_read_t>(&read);
    if (pr) {
        rassert(responses.size() == 1);
        rassert(boost::get<point_read_response_t>(&responses[0].response));
        return responses[0];
    }

    const rget_read_t *rg = boost::get<rget_read_t>(&read);
    if (rg) {
        std::sort(responses.begin(), responses.end(), read_response_cmp);
        rget_read_response_t rg_response;
        rg_response.truncated = false;
        rg_response.key_range = get_region().inner;
        typedef std::vector<read_response_t>::iterator rri_t;

        if (!rg->terminal) {
            rg_response.result = stream_t();
            stream_t *res_stream = boost::get<stream_t>(&rg_response.result);
            for(rri_t i = responses.begin(); i != responses.end(); ++i) {
                // TODO: we're ignoring the limit when recombining.
                const rget_read_response_t *_rr = boost::get<rget_read_response_t>(&i->response);
                rassert(_rr);

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

                res_stream->insert(res_stream->end(), stream->begin(), stream->end());
                rg_response.truncated = rg_response.truncated || _rr->truncated;
            }
        } else if (const Builtin_GroupedMapReduce *gmr = boost::get<Builtin_GroupedMapReduce>(&*rg->terminal)) {
            //Mapreduce
            rg_response.result = groups_t();
            groups_t *res_groups = boost::get<groups_t>(&rg_response.result);
            for(rri_t i = responses.begin(); i != responses.end(); ++i) {
                const rget_read_response_t *_rr = boost::get<rget_read_response_t>(&i->response);
                guarantee(_rr);

                const groups_t *groups = boost::get<groups_t>(&(_rr->result));
                query_language::runtime_environment_t *env = NULL; //obviously this is a problem
                query_language::backtrace_t backtrace;

                for (groups_t::const_iterator j = groups->begin(); j != groups->end(); ++j) {
                    query_language::new_val_scope_t scope(&env->scope);
                    Term base = gmr->reduction().base(),
                         body = gmr->reduction().body();

                    env->scope.put_in_scope(gmr->reduction().var1(), get_with_default(*res_groups, j->first, eval(&base, env, backtrace)));
                    env->scope.put_in_scope(gmr->reduction().var2(), j->second);

                    (*res_groups)[j->first] = eval(&body, env, backtrace);
                }
            }
        } else if (const Reduction *r = boost::get<Reduction>(&*rg->terminal)) {
            rg_response.result = atom_t();
            atom_t *res_atom = boost::get<atom_t>(&rg_response.result);

            query_language::runtime_environment_t *env = NULL; //obviously this is a problem
            query_language::backtrace_t backtrace;

            Term base = r->base();
            *res_atom = eval(&base, env, backtrace);

            for(rri_t i = responses.begin(); i != responses.end(); ++i) {
                const rget_read_response_t *_rr = boost::get<rget_read_response_t>(&i->response);
                guarantee(_rr);

                const atom_t *atom = boost::get<atom_t>(&(_rr->result));

                query_language::new_val_scope_t scope(&env->scope);
                env->scope.put_in_scope(r->var1(), *res_atom);
                env->scope.put_in_scope(r->var2(), *atom);
                Term body = r->body();
                *res_atom = eval(&body, env, backtrace);
            }
        } else if (boost::get<rdb_protocol_details::Length>(&*rg->terminal)) {
        } else if (boost::get<WriteQuery_ForEach>(&*rg->terminal)) {
        }
        return read_response_t(rg_response);
    }

    unreachable("Unknown read response.");
}

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;
{'fraction_non_alphanumeric': 0.12267007603795169, 'fraction_numerical': 0.0012112240091514703, 'mean_word_length': 3.7527982091461465, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2986031', 'n_tokens_mistral': 5189, 'n_tokens_neox': 4973, 'n_words': 830}