<commit_before>#include "BCFKeyValueData.h"
#include "BCFSerialize.h"
#include "diploid.h"
#include "yaml-cpp/yaml.h"
#include "vcf.h"
#include "hfile.h"
#include <sstream>
#include <iomanip>
#include <iostream>
#include <assert.h>
#include <math.h>
#include <thread>
#include <mutex>
#include <sys/time.h>
#include "fcmm.hpp"
#include "khash.h"
#include <regex>
using namespace std;

#define MAX_NUM_CONTIGS_PER_GVCF (10000)
#define MAX_CONTIG_LEN (10000000000)  // 10^10

namespace GLnexus {


// Memory efficient representation of a bucket range. This could
// be turned into a standard C++ iterator, although, that might be
// a bit of an overkill.
class BucketExtent {
private:
    int rid_ = 0;
    int step_ = 0;
    int bgn_ = 0;
    int end_ = 0;
    int current_ = 0;

    // disable copy and assignment constructors
    BucketExtent(const BucketExtent&);
    BucketExtent& operator=(const BucketExtent&);

public:
    BucketExtent(const range &query, int interval_len) {
        rid_ = query.rid;
        step_ = interval_len;
        bgn_ = (query.beg / interval_len) * interval_len;
        end_ = std::max(bgn_, ((query.end-1) / interval_len) * interval_len);
        current_ = 0;
    }

    range begin() {
        range r = range(rid_, bgn_, bgn_ + step_);
        current_ = bgn_;
        return r;
    }

    range next() {
        current_ += step_;
        range r = range(rid_, current_, current_ + step_);
        return r;
    }

    range end() {
        range r = range(rid_, end_, end_ + step_);
        return r;
    }
};

// Map a range into a set of buckets. The records in the range
// can be found by scanning all the buckets. Note that a BCF record
// is placed in a bucket based on its start position.
// It could start in one bucket, and extend into an adjacent bucket(s).
//
// This class separates out the logic for answering the following questions:
//   1) Which buckets should I scan for this query range?
//   2) Which bucket does a bcf1_t with this range go into?
class BCFBucketRange {
private:
    // disable copy and assignment constructors
    BCFBucketRange(const BCFBucketRange&);
    BCFBucketRange& operator=(const BCFBucketRange&);

public:
    static const size_t PREFIX_LENGTH = 24;
    int interval_len;

    // constructor
    BCFBucketRange(int interval_len) : interval_len(interval_len) {};

    // Given the range of a bucket, produce the key prefix for the bucket.
    // Important: the range must be exactly that of the bucket.
    // BCFBucketRange::bucket below translates an arbitrary range into a
    // bucket's range.
    std::string bucket_prefix(const range& rng) {
        stringstream ss;
        // We add leading zeros to ensure that string lexicographic ordering will sort
        // keys in ascending order.
        ss << setw(4) << setfill('0') << rng.rid
           << setw(10) << setfill('0') << rng.beg
           << setw(10) << setfill('0') << rng.end;
        std::string ans = ss.str();
        assert(ans.size() == PREFIX_LENGTH);
        return ans;
    }

    // Produce the complete key for a bucket (given the prefix) in a dataset
    std::string bucket_key(const std::string& prefix, const std::string& dataset) {
        assert(prefix.size() == PREFIX_LENGTH);
        return prefix+dataset;
    }

    // Same as bucket_key(bucket_prefix(rng), dataset)
    // Important: the range must be exactly that of the bucket.
    // BCFBucketRange::bucket below translates an arbitrary range into a
    // bucket's range.
    std::string bucket_key(const range& rng, const std::string& dataset) {
        return bucket_key(bucket_prefix(rng), dataset);
    }

    // Decompose the key into bucket prefix and dataset
    Status parse_key(const string& key, string& bucket, string& dataset) {
        if (key.size() < PREFIX_LENGTH) {
            return Status::Invalid("BCFBucketRange::parse_key: key too small", key);
        }
        bucket = key.substr(0, PREFIX_LENGTH);
        dataset = key.substr(PREFIX_LENGTH);
        return Status::OK();
    }

    // Given arbitrary [query] range, return a structure describing one or
    // more buckets to search through in order to find all records overlapping
    // query. This may be multiple buckets, even for small query ranges, to
    // account for the possibility of records spanning multiple buckets.
    std::shared_ptr<BucketExtent> scan(const range& query) {
        return make_shared<BucketExtent>(query, interval_len);
    }

    // Which bucket should a BCF record be placed in?
    range bucket(bcf1_t *rec) {
        int bgn = (rec->pos / interval_len) * interval_len;
        return range(rec->rid, bgn, bgn + interval_len);
    }
    // ...and the subsequent bucket
    range next_bucket(bcf1_t *rec) {
        int bgn = ((rec->pos / interval_len)+1) * interval_len;
        return range(rec->rid, bgn, bgn + interval_len);
    }
};

size_t BCFKeyValueDataPrefixLength() { return BCFBucketRange::PREFIX_LENGTH; }


// Metadata that is being added to the DB
struct ActiveMetadata {
    set<string> datasets;
    set<string> samples;

    void erase(const string dataset_, set<string> samples_) {
        datasets.erase(dataset_);
        for (const auto& sample : samples_)
            samples.erase(sample);
    }

    void add(const string dataset_, set<string> samples_) {
        datasets.insert(dataset_);
        for (const auto& sample : samples_)
            samples.insert(sample);

    }
};

// std::hash<string> using the string hash function from htslib
class KStringHash {
public:
    std::size_t operator()(string const& s) const  {
        return (size_t) kh_str_hash_func(s.c_str());
    }
};
using BCFHeaderCache = fcmm::Fcmm<string,shared_ptr<const bcf_hdr_t>,hash<string>,KStringHash>;
// this is not a hard limit but the FCMM performance degrades if it's too low
const size_t BCF_HEADER_CACHE_SIZE = 65536;

// pImpl idiom
struct BCFKeyValueData_body {
    KeyValue::DB* db;
    unique_ptr<BCFHeaderCache> header_cache;
    std::unique_ptr<BCFBucketRange> rangeHelper;
    std::mutex mutex;
    ActiveMetadata amd;
    std::mutex statsMutex;
    StatsRangeQuery statsRq; // statistics for range queries
    atomic<size_t> sample_count; // number of samples in the database. could be
                                 // obtained from the size of the current
                                 // all-samples sampleset, but maintained here
                                 // for convenience.
};

auto collections = { "config", "sampleset", "sample_dataset", "header", "bcf" };

BCFKeyValueData::BCFKeyValueData() = default;
BCFKeyValueData::~BCFKeyValueData() = default;

Status BCFKeyValueData::InitializeDB(KeyValue::DB* db,
                                     const vector<pair<string,size_t>>& contigs,
                                     int interval_len) {
    Status s;

    // some basic sanity checks
    if (contigs.size() > MAX_NUM_CONTIGS_PER_GVCF)
        return Status::Invalid("Too many contigs ", std::to_string(contigs.size()));
    for (const auto& p : contigs) {
        size_t contig_len = p.second;
        if (contig_len > MAX_CONTIG_LEN)
            return Status::Invalid("contig is too long ", string(p.first) + " " + std::to_string(contig_len));
    }

    // create collections
    for (const auto& coll : collections) {
        S(db->create_collection(coll));
    }

    KeyValue::CollectionHandle config;
    S(db->collection("config", config));

    // store contigs
    {
        set<string> prev_contigs;
        YAML::Emitter yaml;
        yaml << YAML::BeginSeq;
        for (const auto& p : contigs) {
            if (prev_contigs.find(p.first) != prev_contigs.end()) {
                return Status::Invalid("duplicate reference contig", p.first);
            }
            yaml << YAML::BeginMap;
            yaml << YAML::Key << p.first;
            yaml << YAML::Value << p.second;
            yaml << YAML::EndMap;
        }
        yaml << YAML::EndSeq;
        S(db->put(config, "contigs", yaml.c_str()));
    }

    // store parameters
    {
        YAML::Emitter yaml;
        yaml << YAML::BeginSeq;
        yaml << YAML::BeginMap;
        yaml << YAML::Key << "interval_len";
        yaml << YAML::Value << interval_len;
        yaml << YAML::EndMap;
        yaml << YAML::EndSeq;
        S(db->put(config, "param", yaml.c_str()));
    }

    // create * sample set, with version number 0
    KeyValue::CollectionHandle sampleset;
    S(db->collection("sampleset", sampleset));
    return db->put(sampleset, "*", "0");
}

Status BCFKeyValueData::Open(KeyValue::DB* db, unique_ptr<BCFKeyValueData>& ans) {
    assert(db != nullptr);

    // check database has been initialized
    KeyValue::CollectionHandle coll;
    for (const auto& collnm : collections) {
        if (db->collection(collnm, coll).bad()) {
            return Status::Invalid("database hasn't been properly initialized");
        }
    }

    ans.reset(new BCFKeyValueData());
    ans->body_.reset(new BCFKeyValueData_body);
    ans->body_->db = db;

    // Read the parameters from the DB
    const char *unexpected = "BCFKeyValueData::Open unexpected YAML";
    Status s;
    vector<pair<string,size_t> > param;
    S(ans->body_->db->collection("config", coll));
    try {
        string param_yaml;
        S(ans->body_->db->get(coll, "param", param_yaml));
        YAML::Node n = YAML::Load(param_yaml);
        if (!n.IsSequence()) {
            return Status::Invalid(unexpected, param_yaml);
        }
        for (const auto& item : n) {
            if (!item.IsMap() || item.size() != 1) {
                return Status::Invalid(unexpected, param_yaml);
            }
            auto m = item.as<map<string,size_t>>();
            assert (m.size() == 1);
            param.push_back(*(m.begin()));
        }
    } catch(YAML::Exception& exn) {
        return Status::Invalid("BCFKeyValueData::Open YAML parse error in param", exn.msg);
    }
    if (param.empty()) {
        return Status::Invalid("database has empty parameter metadata");
    }

    // Sift through parameters, sanity check
    int interval_len = -1;
    for (auto item : param) {
        if (item.first == "interval_len") {
            interval_len = item.second;
        }
    }
    if (interval_len <= 0) {
        return Status::Invalid("Corrupt database; bad interval length ", std::to_string(interval_len));
    }

    ans->body_->rangeHelper = make_unique<BCFBucketRange>(interval_len);
    ans->body_->header_cache = make_unique<BCFHeaderCache>(BCF_HEADER_CACHE_SIZE);

    // initialize sample_count
    string sampleset;
    S(ans->all_samples_sampleset(sampleset));
    shared_ptr<const set<string>> all_samples;
    S(ans->sampleset_samples(sampleset, all_samples));
    ans->body_->sample_count = all_samples->size();

    return Status::OK();
}

Status BCFKeyValueData::contigs(vector<pair<string,size_t> >& ans) const {
    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("config", coll));

    // the contigs entry in config contains a YAML list of contigname-size pairs:
    // - 21: 1000000
    // - 22: 1234567
    // ...

    const char *unexpected = "BCFKeyValueData::contigs unexpected YAML";
    ans.clear();
    try {
        string contigs_yaml;
        S(body_->db->get(coll, "contigs", contigs_yaml));
        YAML::Node n = YAML::Load(contigs_yaml);
        if (!n.IsSequence()) {
            return Status::Invalid(unexpected, contigs_yaml);
        }
        for (const auto& item : n) {
            if (!item.IsMap() || item.size() != 1) {
                return Status::Invalid(unexpected, contigs_yaml);
            }
            auto m = item.as<map<string,size_t>>();
            assert (m.size() == 1);
            ans.push_back(*(m.begin()));
        }
    } catch(YAML::Exception& exn) {
        return Status::Invalid("BCFKeyValueData::contigs YAML parse error", exn.msg);
    }
    if (ans.empty()) {
        return Status::Invalid("database has empty contigs metadata");
    }

    return Status::OK();
}

Status BCFKeyValueData::sampleset_samples(const string& sampleset,
                                          shared_ptr<const set<string> >& ans) const {
    if (sampleset == "*") {
        // * is a special reserved sample set representing all available
        // samples in the database. It must be hidden from callers because
        // it's mutable, while sample sets are supposed to be immutable.
        return Status::NotFound();
    }

    // samplesets collection key scheme:
    // sampleset_id
    // sampleset_id\0sample_1
    // sampleset_id\0sample_2
    // ...
    // sampleset_id\0sample_n
    // next_sampleset
    // next_sampleset\0sample_1
    // ...
    // the corresponding values are empty.

    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sampleset",coll));

    unique_ptr<KeyValue::Iterator> it;
    S(body_->db->iterator(coll, sampleset, it));

    if (!it->valid() || it->key_str() != sampleset) {
        return Status::NotFound("sample set not found", sampleset);
    }

    auto samples = make_shared<set<string>>();
    for (s = it->next(); s.ok() && it->valid(); s = it->next()) {
        auto key = it->key_str();
        size_t nullpos = key.find('\0');
        if (nullpos == string::npos || key.substr(0, nullpos) != sampleset) {
            break;
        }
        samples->insert(key.substr(nullpos+1));
    }
    if (s.bad()) return s;

    ans = samples;

    return Status::OK();
}

Status BCFKeyValueData::sample_dataset(const string& sample, string& ans) const {
    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sample_dataset",coll));
    return body_->db->get(coll, sample, ans);
}

Status BCFKeyValueData::all_samples_sampleset(string& ans) {
    Status s;

    // Get the current * sample set version number.
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sampleset",coll));
    unique_ptr<KeyValue::Iterator> it;
    S(body_->db->iterator(coll, "*", it));
    if (!it->valid() || it->key_str() != "*") return Status::NotFound("BCFKeyValueData::all_samples_sampleset: improperly initialized database");
    uint64_t version = strtoull(it->value().first, nullptr, 10);
    ans = "*@" + to_string(version); // this is the desired sample set

    // Does the desired sample set exist already? If so, we are done.
    string ignore;
    s = body_->db->get(coll, ans, ignore);
    if (s.ok() || s != StatusCode::NOT_FOUND) {
        return s;
    }

    // Otherwise, continue to read all the samples in the * sample set, and
    // prepare a write batch creating the desired sample set.
    unique_ptr<KeyValue::WriteBatch> wb;
    S(body_->db->begin_writes(wb));
    S(wb->put(coll, ans, string()));
    for (s = it->next(); s.ok() && it->valid(); s = it->next()) {
        auto key = it->key_str();
        size_t nullpos = key.find('\0');
        if (nullpos == string::npos || key.substr(0, nullpos) != "*") {
            break;
        }
        string sample = key.substr(nullpos+1);
        S(wb->put(coll, ans + string(1,'\0') + sample, string()));
    }
    if (s.bad()) return s;
    it.reset();

    // Commit the new sample set iff no other thread has done so in the<commit_msg>BCFKeyValueData: fix valgrind error revealed by RocksDB upgrade<commit_after>#include "BCFKeyValueData.h"
#include "BCFSerialize.h"
#include "diploid.h"
#include "yaml-cpp/yaml.h"
#include "vcf.h"
#include "hfile.h"
#include <sstream>
#include <iomanip>
#include <iostream>
#include <assert.h>
#include <math.h>
#include <thread>
#include <mutex>
#include <sys/time.h>
#include "fcmm.hpp"
#include "khash.h"
#include <regex>
using namespace std;

#define MAX_NUM_CONTIGS_PER_GVCF (10000)
#define MAX_CONTIG_LEN (10000000000)  // 10^10

namespace GLnexus {


// Memory efficient representation of a bucket range. This could
// be turned into a standard C++ iterator, although, that might be
// a bit of an overkill.
class BucketExtent {
private:
    int rid_ = 0;
    int step_ = 0;
    int bgn_ = 0;
    int end_ = 0;
    int current_ = 0;

    // disable copy and assignment constructors
    BucketExtent(const BucketExtent&);
    BucketExtent& operator=(const BucketExtent&);

public:
    BucketExtent(const range &query, int interval_len) {
        rid_ = query.rid;
        step_ = interval_len;
        bgn_ = (query.beg / interval_len) * interval_len;
        end_ = std::max(bgn_, ((query.end-1) / interval_len) * interval_len);
        current_ = 0;
    }

    range begin() {
        range r = range(rid_, bgn_, bgn_ + step_);
        current_ = bgn_;
        return r;
    }

    range next() {
        current_ += step_;
        range r = range(rid_, current_, current_ + step_);
        return r;
    }

    range end() {
        range r = range(rid_, end_, end_ + step_);
        return r;
    }
};

// Map a range into a set of buckets. The records in the range
// can be found by scanning all the buckets. Note that a BCF record
// is placed in a bucket based on its start position.
// It could start in one bucket, and extend into an adjacent bucket(s).
//
// This class separates out the logic for answering the following questions:
//   1) Which buckets should I scan for this query range?
//   2) Which bucket does a bcf1_t with this range go into?
class BCFBucketRange {
private:
    // disable copy and assignment constructors
    BCFBucketRange(const BCFBucketRange&);
    BCFBucketRange& operator=(const BCFBucketRange&);

public:
    static const size_t PREFIX_LENGTH = 24;
    int interval_len;

    // constructor
    BCFBucketRange(int interval_len) : interval_len(interval_len) {};

    // Given the range of a bucket, produce the key prefix for the bucket.
    // Important: the range must be exactly that of the bucket.
    // BCFBucketRange::bucket below translates an arbitrary range into a
    // bucket's range.
    std::string bucket_prefix(const range& rng) {
        stringstream ss;
        // We add leading zeros to ensure that string lexicographic ordering will sort
        // keys in ascending order.
        ss << setw(4) << setfill('0') << rng.rid
           << setw(10) << setfill('0') << rng.beg
           << setw(10) << setfill('0') << rng.end;
        std::string ans = ss.str();
        assert(ans.size() == PREFIX_LENGTH);
        return ans;
    }

    // Produce the complete key for a bucket (given the prefix) in a dataset
    std::string bucket_key(const std::string& prefix, const std::string& dataset) {
        assert(prefix.size() == PREFIX_LENGTH);
        return prefix+dataset;
    }

    // Same as bucket_key(bucket_prefix(rng), dataset)
    // Important: the range must be exactly that of the bucket.
    // BCFBucketRange::bucket below translates an arbitrary range into a
    // bucket's range.
    std::string bucket_key(const range& rng, const std::string& dataset) {
        return bucket_key(bucket_prefix(rng), dataset);
    }

    // Decompose the key into bucket prefix and dataset
    Status parse_key(const string& key, string& bucket, string& dataset) {
        if (key.size() < PREFIX_LENGTH) {
            return Status::Invalid("BCFBucketRange::parse_key: key too small", key);
        }
        bucket = key.substr(0, PREFIX_LENGTH);
        dataset = key.substr(PREFIX_LENGTH);
        return Status::OK();
    }

    // Given arbitrary [query] range, return a structure describing one or
    // more buckets to search through in order to find all records overlapping
    // query. This may be multiple buckets, even for small query ranges, to
    // account for the possibility of records spanning multiple buckets.
    std::shared_ptr<BucketExtent> scan(const range& query) {
        return make_shared<BucketExtent>(query, interval_len);
    }

    // Which bucket should a BCF record be placed in?
    range bucket(bcf1_t *rec) {
        int bgn = (rec->pos / interval_len) * interval_len;
        return range(rec->rid, bgn, bgn + interval_len);
    }
    // ...and the subsequent bucket
    range next_bucket(bcf1_t *rec) {
        int bgn = ((rec->pos / interval_len)+1) * interval_len;
        return range(rec->rid, bgn, bgn + interval_len);
    }
};

size_t BCFKeyValueDataPrefixLength() { return BCFBucketRange::PREFIX_LENGTH; }


// Metadata that is being added to the DB
struct ActiveMetadata {
    set<string> datasets;
    set<string> samples;

    void erase(const string dataset_, set<string> samples_) {
        datasets.erase(dataset_);
        for (const auto& sample : samples_)
            samples.erase(sample);
    }

    void add(const string dataset_, set<string> samples_) {
        datasets.insert(dataset_);
        for (const auto& sample : samples_)
            samples.insert(sample);

    }
};

// std::hash<string> using the string hash function from htslib
class KStringHash {
public:
    std::size_t operator()(string const& s) const  {
        return (size_t) kh_str_hash_func(s.c_str());
    }
};
using BCFHeaderCache = fcmm::Fcmm<string,shared_ptr<const bcf_hdr_t>,hash<string>,KStringHash>;
// this is not a hard limit but the FCMM performance degrades if it's too low
const size_t BCF_HEADER_CACHE_SIZE = 65536;

// pImpl idiom
struct BCFKeyValueData_body {
    KeyValue::DB* db;
    unique_ptr<BCFHeaderCache> header_cache;
    std::unique_ptr<BCFBucketRange> rangeHelper;
    std::mutex mutex;
    ActiveMetadata amd;
    std::mutex statsMutex;
    StatsRangeQuery statsRq; // statistics for range queries
    atomic<size_t> sample_count; // number of samples in the database. could be
                                 // obtained from the size of the current
                                 // all-samples sampleset, but maintained here
                                 // for convenience.
};

auto collections = { "config", "sampleset", "sample_dataset", "header", "bcf" };

BCFKeyValueData::BCFKeyValueData() = default;
BCFKeyValueData::~BCFKeyValueData() = default;

Status BCFKeyValueData::InitializeDB(KeyValue::DB* db,
                                     const vector<pair<string,size_t>>& contigs,
                                     int interval_len) {
    Status s;

    // some basic sanity checks
    if (contigs.size() > MAX_NUM_CONTIGS_PER_GVCF)
        return Status::Invalid("Too many contigs ", std::to_string(contigs.size()));
    for (const auto& p : contigs) {
        size_t contig_len = p.second;
        if (contig_len > MAX_CONTIG_LEN)
            return Status::Invalid("contig is too long ", string(p.first) + " " + std::to_string(contig_len));
    }

    // create collections
    for (const auto& coll : collections) {
        S(db->create_collection(coll));
    }

    KeyValue::CollectionHandle config;
    S(db->collection("config", config));

    // store contigs
    {
        set<string> prev_contigs;
        YAML::Emitter yaml;
        yaml << YAML::BeginSeq;
        for (const auto& p : contigs) {
            if (prev_contigs.find(p.first) != prev_contigs.end()) {
                return Status::Invalid("duplicate reference contig", p.first);
            }
            yaml << YAML::BeginMap;
            yaml << YAML::Key << p.first;
            yaml << YAML::Value << p.second;
            yaml << YAML::EndMap;
        }
        yaml << YAML::EndSeq;
        S(db->put(config, "contigs", yaml.c_str()));
    }

    // store parameters
    {
        YAML::Emitter yaml;
        yaml << YAML::BeginSeq;
        yaml << YAML::BeginMap;
        yaml << YAML::Key << "interval_len";
        yaml << YAML::Value << interval_len;
        yaml << YAML::EndMap;
        yaml << YAML::EndSeq;
        S(db->put(config, "param", yaml.c_str()));
    }

    // create * sample set, with version number 0
    KeyValue::CollectionHandle sampleset;
    S(db->collection("sampleset", sampleset));
    return db->put(sampleset, "*", "0");
}

Status BCFKeyValueData::Open(KeyValue::DB* db, unique_ptr<BCFKeyValueData>& ans) {
    assert(db != nullptr);

    // check database has been initialized
    KeyValue::CollectionHandle coll;
    for (const auto& collnm : collections) {
        if (db->collection(collnm, coll).bad()) {
            return Status::Invalid("database hasn't been properly initialized");
        }
    }

    ans.reset(new BCFKeyValueData());
    ans->body_.reset(new BCFKeyValueData_body);
    ans->body_->db = db;

    // Read the parameters from the DB
    const char *unexpected = "BCFKeyValueData::Open unexpected YAML";
    Status s;
    vector<pair<string,size_t> > param;
    S(ans->body_->db->collection("config", coll));
    try {
        string param_yaml;
        S(ans->body_->db->get(coll, "param", param_yaml));
        YAML::Node n = YAML::Load(param_yaml);
        if (!n.IsSequence()) {
            return Status::Invalid(unexpected, param_yaml);
        }
        for (const auto& item : n) {
            if (!item.IsMap() || item.size() != 1) {
                return Status::Invalid(unexpected, param_yaml);
            }
            auto m = item.as<map<string,size_t>>();
            assert (m.size() == 1);
            param.push_back(*(m.begin()));
        }
    } catch(YAML::Exception& exn) {
        return Status::Invalid("BCFKeyValueData::Open YAML parse error in param", exn.msg);
    }
    if (param.empty()) {
        return Status::Invalid("database has empty parameter metadata");
    }

    // Sift through parameters, sanity check
    int interval_len = -1;
    for (auto item : param) {
        if (item.first == "interval_len") {
            interval_len = item.second;
        }
    }
    if (interval_len <= 0) {
        return Status::Invalid("Corrupt database; bad interval length ", std::to_string(interval_len));
    }

    ans->body_->rangeHelper = make_unique<BCFBucketRange>(interval_len);
    ans->body_->header_cache = make_unique<BCFHeaderCache>(BCF_HEADER_CACHE_SIZE);

    // initialize sample_count
    string sampleset;
    S(ans->all_samples_sampleset(sampleset));
    shared_ptr<const set<string>> all_samples;
    S(ans->sampleset_samples(sampleset, all_samples));
    ans->body_->sample_count = all_samples->size();

    return Status::OK();
}

Status BCFKeyValueData::contigs(vector<pair<string,size_t> >& ans) const {
    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("config", coll));

    // the contigs entry in config contains a YAML list of contigname-size pairs:
    // - 21: 1000000
    // - 22: 1234567
    // ...

    const char *unexpected = "BCFKeyValueData::contigs unexpected YAML";
    ans.clear();
    try {
        string contigs_yaml;
        S(body_->db->get(coll, "contigs", contigs_yaml));
        YAML::Node n = YAML::Load(contigs_yaml);
        if (!n.IsSequence()) {
            return Status::Invalid(unexpected, contigs_yaml);
        }
        for (const auto& item : n) {
            if (!item.IsMap() || item.size() != 1) {
                return Status::Invalid(unexpected, contigs_yaml);
            }
            auto m = item.as<map<string,size_t>>();
            assert (m.size() == 1);
            ans.push_back(*(m.begin()));
        }
    } catch(YAML::Exception& exn) {
        return Status::Invalid("BCFKeyValueData::contigs YAML parse error", exn.msg);
    }
    if (ans.empty()) {
        return Status::Invalid("database has empty contigs metadata");
    }

    return Status::OK();
}

Status BCFKeyValueData::sampleset_samples(const string& sampleset,
                                          shared_ptr<const set<string> >& ans) const {
    if (sampleset == "*") {
        // * is a special reserved sample set representing all available
        // samples in the database. It must be hidden from callers because
        // it's mutable, while sample sets are supposed to be immutable.
        return Status::NotFound();
    }

    // samplesets collection key scheme:
    // sampleset_id
    // sampleset_id\0sample_1
    // sampleset_id\0sample_2
    // ...
    // sampleset_id\0sample_n
    // next_sampleset
    // next_sampleset\0sample_1
    // ...
    // the corresponding values are empty.

    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sampleset",coll));

    unique_ptr<KeyValue::Iterator> it;
    S(body_->db->iterator(coll, sampleset, it));

    if (!it->valid() || it->key_str() != sampleset) {
        return Status::NotFound("sample set not found", sampleset);
    }

    auto samples = make_shared<set<string>>();
    for (s = it->next(); s.ok() && it->valid(); s = it->next()) {
        auto key = it->key_str();
        size_t nullpos = key.find('\0');
        if (nullpos == string::npos || key.substr(0, nullpos) != sampleset) {
            break;
        }
        samples->insert(key.substr(nullpos+1));
    }
    if (s.bad()) return s;

    ans = samples;

    return Status::OK();
}

Status BCFKeyValueData::sample_dataset(const string& sample, string& ans) const {
    Status s;
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sample_dataset",coll));
    return body_->db->get(coll, sample, ans);
}

Status BCFKeyValueData::all_samples_sampleset(string& ans) {
    Status s;

    // Get the current * sample set version number.
    KeyValue::CollectionHandle coll;
    S(body_->db->collection("sampleset",coll));
    unique_ptr<KeyValue::Iterator> it;
    S(body_->db->iterator(coll, "*", it));
    if (!it->valid() || it->key_str() != "*") return Status::NotFound("BCFKeyValueData::all_samples_sampleset: improperly initialized database");
    uint64_t version = strtoull(string(it->value().first, it->value().second).c_str(), nullptr, 10);
    ans = "*@" + to_string(version); // this is the desired sample set

    // Does the desired sample set exist already? If so, we are done.
    string ignore;
    s = body_->db->get(coll, ans, ignore);
    if (s.ok() || s != StatusCode::NOT_FOUND) {
        return s;
    }

    // Otherwise, continue to read all the samples in the * sample set, and
    // prepare a write batch creating the desired sample set.
    unique_ptr<KeyValue::WriteBatch> wb;
    S(body_->db->begin_writes(wb));
    S(wb->put(coll, ans, string()));
    for (s = it->next(); s.ok() && it->valid(); s = it->next()) {
        auto key = it->key_str();
        size_t nullpos = key.find('\0');
        if (nullpos == string::npos || key.substr(0, nullpos) != "*") {
            break;
        }
        string sample = key.substr(nullpos+1);
        S(wb->put(coll, ans + string(1,'\0') + sample, string()));
    }
    if (s.bad()) return s;
    it.reset();

    // Commit the new sample set iff no other thread has done so in the
{'fraction_non_alphanumeric': 0.1303864513175963, 'fraction_numerical': 0.006146603020989996, 'mean_word_length': 2.6683857040057566, 'pattern_counts': {'":': 0, '<': 197, '<?xml version=': 0, '>': 228, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '774846', 'n_tokens_mistral': 9549, 'n_tokens_neox': 9208, 'n_words': 2547}