csv_reader.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <algorithm>
#include <deque>
#include <exception>
#include <fstream>
#include <functional>
#include <iterator>
#include <memory>
#include <sstream>
#include <string>
#include <vector>
 
#if !defined(CSV_ENABLE_THREADS) || CSV_ENABLE_THREADS
#include <mutex>
#include <thread>
#endif
 
#include "basic_csv_parser.hpp"
#include "common.hpp"
#include "data_type.hpp"
#include "csv_format.hpp"
 
namespace csv {
#if CSV_ENABLE_THREADS
    inline void join_worker(std::thread& worker) {
        if (worker.joinable()) worker.join();
    }
 
    #define JOIN_WORKER(worker) join_worker(worker)
#else
    #define JOIN_WORKER(worker) ((void)0)
#endif
 
    class CSVReader {
    public:
        class iterator {
        public:
            #ifndef DOXYGEN_SHOULD_SKIP_THIS
            using value_type = CSVRow;
            using difference_type = std::ptrdiff_t;
            using pointer = CSVRow * ;
            using reference = CSVRow & ;
            using iterator_category = std::input_iterator_tag;
            #endif
 
            iterator() = default;
            iterator(CSVReader* reader) : daddy(reader) {}
            iterator(CSVReader*, CSVRow&&);
 
            CONSTEXPR_14 reference operator*() { return this->row; }
            CONSTEXPR_14 reference operator*() const { return const_cast<reference>(this->row); }
 
            CONSTEXPR_14 pointer operator->() { return &(this->row); }
            CONSTEXPR_14 pointer operator->() const { return const_cast<pointer>(&(this->row)); }
 
            iterator& operator++();   
            iterator operator++(int); 
            CONSTEXPR bool operator==(const iterator& other) const noexcept {
                return (this->daddy == other.daddy) && (this->i == other.i);
            }
 
            CONSTEXPR bool operator!=(const iterator& other) const noexcept { return !operator==(other); }
        private:
            CSVReader * daddy = nullptr;  // Pointer to parent
            CSVRow row;                   // Current row
            size_t i = 0;               // Index of current row
        };
 
 
        CSVReader(csv::string_view filename, const CSVFormat& format = CSVFormat::guess_csv()) : _format(format) {
#if defined(__EMSCRIPTEN__)
            this->owned_stream = std::unique_ptr<std::istream>(
                new std::ifstream(std::string(filename), std::ios::binary)
            );
 
            if (!(*this->owned_stream)) {
                throw std::runtime_error("Cannot open file " + std::string(filename));
            }
 
            this->init_from_stream(*this->owned_stream, format);
#else
            // C4316: MmapParser may carry over-aligned SIMD members. Allocation
            // alignment is handled by the allocator on supported platforms;
            // suppress MSVC's false-positive warning at this site.
            CSV_MSVC_PUSH_DISABLE(4316)
            this->init_parser(std::unique_ptr<internals::IBasicCSVParser>(
                new internals::MmapParser(filename, format, this->col_names)
            ));
            CSV_MSVC_POP
#endif
        }
 
        template<typename TStream,
            csv::enable_if_t<std::is_base_of<std::istream, TStream>::value, int> = 0>
        CSVReader(TStream &source, CSVFormat format = CSVFormat::guess_csv()) : _format(format) {
            this->init_from_stream(source, format);
        }
 
        CSVReader(std::unique_ptr<std::istream> source,
            const CSVFormat& format = CSVFormat::guess_csv()) : _format(format), owned_stream(std::move(source)) {
            if (!this->owned_stream) {
                throw std::invalid_argument("CSVReader requires a non-null stream");
            }
 
            this->init_from_stream(*this->owned_stream, format);
        }
 
        CSVReader(const CSVReader&) = delete;             
        CSVReader& operator=(const CSVReader&) = delete;  
 
        CSVReader(CSVReader&& other) noexcept :
            _format(std::move(other._format)),
            col_names(std::move(other.col_names)),
            parser(std::move(other.parser)),
            records(std::move(other.records)),
            owned_stream(std::move(other.owned_stream)),
            n_cols(other.n_cols),
            _n_rows(other._n_rows),
            header_trimmed(other.header_trimmed),
            _chunk_size(other._chunk_size),
            _read_requested(other._read_requested),
            read_csv_exception(other.take_read_csv_exception()) {
            JOIN_WORKER(other.read_csv_worker);
 
            other.n_cols = 0;
            other._n_rows = 0;
            other.header_trimmed = false;
            other._read_requested = false;
            other._chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT;
        }
 
        CSVReader& operator=(CSVReader&& other) noexcept {
            if (this == &other) {
                return *this;
            }
 
            JOIN_WORKER(this->read_csv_worker);
            JOIN_WORKER(other.read_csv_worker);
 
            this->_format = std::move(other._format);
            this->col_names = std::move(other.col_names);
            this->parser = std::move(other.parser);
            this->records = std::move(other.records);
            this->owned_stream = std::move(other.owned_stream);
            this->n_cols = other.n_cols;
            this->_n_rows = other._n_rows;
            this->header_trimmed = other.header_trimmed;
            this->_chunk_size = other._chunk_size;
            this->_read_requested = other._read_requested;
            this->read_csv_exception = other.take_read_csv_exception();
 
            other.n_cols = 0;
            other._n_rows = 0;
            other.header_trimmed = false;
            other._read_requested = false;
            other._chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT;
 
            return *this;
        }
 
        ~CSVReader() {
            JOIN_WORKER(this->read_csv_worker);
        }
 
        bool read_row(CSVRow &row);
        iterator begin();
        CSV_CONST iterator end() const noexcept;
 
        bool eof() const noexcept { return this->parser->eof(); }
 
        CSVFormat get_format() const;
        std::vector<std::string> get_col_names() const;
        int index_of(csv::string_view col_name) const;
 
 
        CONSTEXPR bool empty() const noexcept { return this->n_rows() == 0; }
 
        CONSTEXPR size_t n_rows() const noexcept { return this->_n_rows; }
 
        bool utf8_bom() const noexcept { return this->parser->utf8_bom(); }
 
    protected:
        void set_col_names(const std::vector<std::string>&);
 
        CSVFormat _format;
 
 
        internals::ColNamesPtr col_names = std::make_shared<internals::ColNames>();
 
        std::unique_ptr<internals::IBasicCSVParser> parser = nullptr;
 
        std::unique_ptr<RowCollection> records{new RowCollection(100)};
 
        std::unique_ptr<std::istream> owned_stream = nullptr;
 
        size_t n_cols = 0;  
        size_t _n_rows = 0; 
        bool read_csv(size_t bytes = internals::CSV_CHUNK_SIZE_DEFAULT);
 
    private:
        bool header_trimmed = false;
    #if CSV_ENABLE_THREADS
        std::thread read_csv_worker; 
    #endif
        size_t _chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT; 
        bool _read_requested = false; 
 
        std::exception_ptr read_csv_exception = nullptr;
#if CSV_ENABLE_THREADS
        std::mutex read_csv_exception_lock;
#endif
 
        void set_read_csv_exception(std::exception_ptr eptr) {
#if CSV_ENABLE_THREADS
            std::lock_guard<std::mutex> lock(this->read_csv_exception_lock);
#endif
            this->read_csv_exception = std::move(eptr);
        }
 
        std::exception_ptr take_read_csv_exception() {
#if CSV_ENABLE_THREADS
            std::lock_guard<std::mutex> lock(this->read_csv_exception_lock);
#endif
            auto eptr = this->read_csv_exception;
            this->read_csv_exception = nullptr;
            return eptr;
        }
 
        void rethrow_read_csv_exception_if_any() {
            if (auto eptr = this->take_read_csv_exception()) {
                std::rethrow_exception(eptr);
            }
        }
 
        void init_parser(std::unique_ptr<internals::IBasicCSVParser> parser);
 
        template<typename TStream,
            csv::enable_if_t<std::is_base_of<std::istream, TStream>::value, int> = 0>
        void init_from_stream(TStream& source, CSVFormat format) {
            // C4316: StreamParser may have over-aligned SIMD members; heap allocation
            // alignment is handled correctly at runtime via the allocator on supported
            // platforms. Suppress the MSVC false-positive here.
            CSV_MSVC_PUSH_DISABLE(4316)
            this->init_parser(
                std::unique_ptr<internals::IBasicCSVParser>(
                    new internals::StreamParser<TStream>(source, format, this->col_names)
                )
            );
            CSV_MSVC_POP
        }
 
        void initial_read() {
#if CSV_ENABLE_THREADS
            this->read_csv_worker = std::thread(&CSVReader::read_csv, this, this->_chunk_size);
            this->read_csv_worker.join();
#else
            this->read_csv(this->_chunk_size);
#endif
            this->rethrow_read_csv_exception_if_any();
        }
 
        void trim_header();
    };
}