Method bulkInsert

src/storage/rocksdb.cpp:63–116

}

void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
{
    if (celem >= 16384) {
        rocksdb::Options options = DefaultRocksDBOptions();
        rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
        std::string file_path = m_pfactory->getTempFolder() + "/tmpIngest.";
        file_path += std::to_string(gettid());
        file_path += ".sst";

        rocksdb::Status s = sst_file_writer.Open(file_path);
        if (!s.ok())
            goto LFallback;

        // Insert rows into the SST file, note that inserted keys must be 
        // strictly increasing (based on options.comparator)
        for (size_t ielem = 0; ielem < celem; ++ielem) {
            std::string prefixed_key = prefixKey(rgkeys[ielem], rgcbkeys[ielem]);
            s = sst_file_writer.Put(rocksdb::Slice(prefixed_key), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
            if (!s.ok()) {
                unlink(file_path.c_str());
                goto LFallback;
            }
        }

        s = sst_file_writer.Finish();
        if (!s.ok()) {
            unlink(file_path.c_str());
            goto LFallback;
        }

        auto ingestOptions = rocksdb::IngestExternalFileOptions();
        ingestOptions.move_files = true;
        ingestOptions.write_global_seqno = false;
        ingestOptions.failed_move_fall_back_to_copy = false;

        // Ingest the external SST file into the DB
        s = m_spdb->IngestExternalFile(m_spcolfamily.get(), {file_path}, ingestOptions);
        if (!s.ok()) {
            unlink(file_path.c_str());
            goto LFallback;
        }
    } else {
    LFallback:
        auto spbatch = std::make_unique<rocksdb::WriteBatch>();
        for (size_t ielem = 0; ielem < celem; ++ielem) {
            std::string prefixed_key = prefixKey(rgkeys[ielem], rgcbkeys[ielem]);
            spbatch->Put(m_spcolfamily.get(), rocksdb::Slice(prefixed_key), rocksdb::Slice(rgvals[ielem], rgcbvals[ielem]));
        }
        m_spdb->Write(WriteOptions(), spbatch.get());
    }

    std::unique_lock<fastlock> l(m_lock);
    m_count += celem;
}

bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
{