GPUdbIngestor.hpp

Go to the documentation of this file.

#ifndef __GPUDB_INGESTOR_HPP__
#define __GPUDB_INGESTOR_HPP__

#include "gpudb/GPUdb.hpp"
#include "gpudb/Http.hpp"
#include "gpudb/Type.hpp"
#include "gpudb/utils/GPUdbMultiHeadIOUtils.h"


#include <atomic>
#include <map>
#include <string>
#include <vector>
#include <mutex>

#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>


namespace gpudb
{

// Forward declaration
class GPUdb;


/*
 * The multi-head ingestor class (also handles regular insertion).  Using this class is
 * significantly more computation-intensive compared to a regular insertion.  So, it is
 * highly recommended to use this ingestor only if multi-head ingestion is actually turned
 * on in the server and there is a large volume of records to be inserted.
 */
class GPUdbIngestor : private boost::noncopyable
{

public:

    GPUdbIngestor( const gpudb::GPUdb& db, const gpudb::Type& record_type,
                   const std::string& table_name,
                   const WorkerList& worker_list,
                   const std::map<std::string, std::string>& insert_options,
                   size_t batch_size );

    GPUdbIngestor( const gpudb::GPUdb& db, const gpudb::Type& record_type,
                   const std::string& table_name,
                   const WorkerList& worker_list,
                   size_t batch_size );


    GPUdbIngestor( const gpudb::GPUdb& db, const gpudb::Type& record_type,
                   const std::string& table_name,
                   const std::map<std::string, std::string>& insert_options,
                   size_t batch_size );

    GPUdbIngestor( const gpudb::GPUdb& db, const gpudb::Type& record_type,
                   const std::string& table_name, size_t batch_size );

    ~GPUdbIngestor();

    const std::string& getTableName() const { return m_table_name; }

    const gpudb::GPUdb& getGPUdb() const { return m_db; }

    const std::map<std::string, std::string>& getOptions() const { return m_insert_options; }

    size_t getCountInserted() const { return m_count_inserted; }

    size_t getCountUpdated() const { return m_count_updated; }

    std::vector<GPUdbInsertionException> getErrors();

    std::vector<GPUdbInsertionException> getWarnings();

    void flush();


    void insert( gpudb::GenericRecord record );


    void insert( std::vector<gpudb::GenericRecord> records );


private:

    typedef std::map<std::string, std::string>          str_to_str_map_t;
    typedef boost::shared_ptr<gpudb::WorkerQueue>       worker_queue_ptr_t;


    GPUdbIngestor();

    void construct( const gpudb::GPUdb& db,
                    const gpudb::Type& record_type,
                    const std::string& table_name,
                    const WorkerList& worker_list,
                    size_t batch_size );

    /*
     * Insert the given list of records to the database residing at the given URL.
     * Upon any error, thrown InsertException with the queue of records passed into it.
     */
    void flush( const std::vector<gpudb::GenericRecord>& queue,
                const gpudb::HttpUrl& url );

    const gpudb::GPUdb&              m_db;
    std::string                      m_table_name;
    size_t                           m_batch_size;
    bool                             m_return_individual_errors;
    bool                             m_simulate_error_mode; // Simulate returnIndividualErrors after an error
    std::atomic<size_t>              m_count_inserted;
    std::atomic<size_t>              m_count_updated;
    str_to_str_map_t                 m_insert_options;
    gpudb::Type                      m_record_type;
    gpudb::RecordKeyBuilder*         m_primary_key_builder_ptr;
    gpudb::RecordKeyBuilder*         m_shard_key_builder_ptr;
//    record_key_buildter_ptr          m_primary_key_builder_ptr;
//    record_key_buildter_ptr          m_shard_key_builder_ptr;
    std::vector<int32_t>             m_routing_table;
    std::vector<worker_queue_ptr_t>  m_worker_queues;
    std::vector<GPUdbInsertionException> m_error_list;
    std::vector<GPUdbInsertionException> m_warning_list;
    std::mutex                       m_error_list_lock;

};  // end class GPUdbIngestor



} // namespace gpudb



#endif // __GPUDB_INGESTOR_HPP__