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>
 
#include "common.hpp"
#include "csv_exceptions.hpp"
#include "data_type.hpp"
#include "csv_format.hpp"
#include "parser/mmap.hpp"
#include "parser/scheduler.hpp"
#include "parser/stream.hpp"
 
namespace csv {
    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),
              read_scheduler_(format.is_threading_enabled()) {
#if defined(__EMSCRIPTEN__)
            this->owned_stream = std::unique_ptr<std::istream>(
                new std::ifstream(std::string(filename), std::ios::binary)
            );
 
            if (!(*this->owned_stream)) {
                internals::throw_cannot_open_file(filename);
            }
 
            this->init_from_stream(*this->owned_stream, format);
#else
            this->init_parser(std::unique_ptr<internals::parser::CSVParserDriverBase>(
                new internals::parser::MmapParser(filename, format, this->col_names)
            ));
#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),
              read_scheduler_(format.is_threading_enabled()) {
            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)),
              read_scheduler_(format.is_threading_enabled()) {
            if (!this->owned_stream) {
                throw std::invalid_argument(internals::ERROR_READER_NULL_STREAM);
            }
 
            this->init_from_stream(*this->owned_stream, format);
        }
 
        CSVReader(const CSVReader&) = delete;             
        CSVReader& operator=(const CSVReader&) = delete;  
 
        CSVReader(CSVReader&& other) noexcept :
            read_scheduler_(other._format.is_threading_enabled()) {
            other.read_scheduler_.join();
            this->move_state_from(other);
        }
 
        CSVReader& operator=(CSVReader&& other) noexcept {
            if (this == &other) {
                return *this;
            }
 
            this->read_scheduler_.join();
            other.read_scheduler_.join();
            this->move_state_from(other);
 
            return *this;
        }
 
        ~CSVReader() {
            this->read_scheduler_.join();
        }
 
 
        bool read_row(CSVRow &row);
 
        bool read_chunk(std::vector<CSVRow>& out, size_t max_rows);
        iterator begin();
        CSV_CONST iterator end() const noexcept;
 
        bool eof() const noexcept { return this->parser->eof(); }
 
 
        CSVFormat get_format() const;
 
        const std::vector<std::string>& get_col_names() const{
            static const std::vector<std::string> empty_col_names;
            return (this->col_names) ? this->col_names->get_col_names() : empty_col_names;
        }
 
        internals::ConstColNamesPtr col_names_ptr() const noexcept {
            return this->col_names;
        }
 
        int index_of(csv::string_view col_name) const {
            return this->col_names->index_of(col_name);
        }
 
 
        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(); }
 
        internals::SpeculativeParseDiagnostics speculative_diagnostics() const noexcept {
            return this->parser ? this->parser->speculative_diagnostics()
                : internals::SpeculativeParseDiagnostics();
        }
 
        size_t parse_worker_count() const noexcept {
            return this->parser ? this->parser->parse_worker_count() : 1;
        }
 
    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::parser::CSVParserDriverBase> 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;
        
        size_t _chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT; 
        bool _read_requested = false; 
        internals::CSVReadScheduler read_scheduler_;
 
        void move_state_from(CSVReader& other) noexcept {
            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_scheduler_.set_threading_enabled(this->_format.is_threading_enabled());
            this->read_scheduler_.adopt_exception(other.read_scheduler_.take_exception());
            other.reset_after_move();
        }
 
        void reset_after_move() noexcept {
            this->n_cols = 0;
            this->_n_rows = 0;
            this->header_trimmed = false;
            this->_read_requested = false;
            this->_chunk_size = internals::CSV_CHUNK_SIZE_DEFAULT;
        }
 
 
        void init_parser(std::unique_ptr<internals::parser::CSVParserDriverBase> 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) {
            this->init_parser(
                std::unique_ptr<internals::parser::CSVParserDriverBase>(
                    new internals::parser::StreamParser<TStream>(source, format, this->col_names)
                )
            );
        }
 
        void initial_read() {
            this->read_scheduler_.run([this] { this->read_csv(this->_chunk_size); });
            this->read_scheduler_.join();
            this->read_scheduler_.rethrow_exception_if_any();
        }
 
        void trim_header();
 
 
        bool accept_row(CSVRow&& candidate, CSVRow* single_row, std::vector<CSVRow>* batch_rows);
 
        bool check_for_rows();
 
        void drain_rows_into_chunk(std::vector<CSVRow>& out, size_t max_rows);
    };
 
    #ifdef CSV_HAS_CXX20
    static_assert(
        internals::csv_write_rows_input_range<CSVReader>,
        "CSVReader must remain compatible with csv::DelimWriter::write_rows()."
    );
    #endif
}