basic_csv_parser.hpp

Go to the documentation of this file.

 
#pragma once
#include <algorithm>
#include <array>
#include <fstream>
#include <memory>
#include <unordered_map>
#include <unordered_set>
#include <thread>
#include <vector>
 
#include "../external/mio.hpp"
#include "col_names.hpp"
#include "common.hpp"
#include "csv_format.hpp"
#include "csv_row.hpp"
#include "thread_safe_deque.hpp"
 
namespace csv {
    namespace internals {
        constexpr const int UNINITIALIZED_FIELD = -1;
 
        template<typename OutArray, typename T = typename OutArray::type>
        CSV_CONST CONSTEXPR_17 OutArray arrayToDefault(T&& value)
        {
            OutArray a {};
            for (auto& e : a)
                 e = value;
            return a;
        }
 
        CSV_CONST CONSTEXPR_17 ParseFlagMap make_parse_flags(char delimiter) {
            auto ret = arrayToDefault<ParseFlagMap>(ParseFlags::NOT_SPECIAL);
            ret[delimiter + CHAR_OFFSET] = ParseFlags::DELIMITER;
            ret['\r' + CHAR_OFFSET] = ParseFlags::NEWLINE;
            ret['\n' + CHAR_OFFSET] = ParseFlags::NEWLINE;
            return ret;
        }
 
        CSV_CONST CONSTEXPR_17 ParseFlagMap make_parse_flags(char delimiter, char quote_char) {
            std::array<ParseFlags, 256> ret = make_parse_flags(delimiter);
            ret[quote_char + CHAR_OFFSET] = ParseFlags::QUOTE;
            return ret;
        }
 
        CSV_CONST CONSTEXPR_17 WhitespaceMap make_ws_flags(const char* ws_chars, size_t n_chars) {
            auto ret = arrayToDefault<WhitespaceMap>(false);
            for (size_t j = 0; j < n_chars; j++) {
                ret[ws_chars[j] + CHAR_OFFSET] = true;
            }
            return ret;
        }
 
        inline WhitespaceMap make_ws_flags(const std::vector<char>& flags) {
            return make_ws_flags(flags.data(), flags.size());
        }
 
        CSV_INLINE size_t get_file_size(csv::string_view filename);
 
        CSV_INLINE std::string get_csv_head(csv::string_view filename);
    }
 
    using RowCollection = internals::ThreadSafeDeque<CSVRow>;
 
    namespace internals {
        class IBasicCSVParser {
        public:
            IBasicCSVParser() = default;
            IBasicCSVParser(const CSVFormat&, const ColNamesPtr&);
            IBasicCSVParser(const ParseFlagMap& parse_flags, const WhitespaceMap& ws_flags
            ) : _parse_flags(parse_flags), _ws_flags(ws_flags) {}
 
            virtual ~IBasicCSVParser() {}
 
            bool eof() { return this->_eof; }
 
            virtual void next(size_t bytes) = 0;
 
            void end_feed();
 
            CONSTEXPR_17 ParseFlags parse_flag(const char ch) const noexcept {
                return _parse_flags.data()[ch + CHAR_OFFSET];
            }
 
            CONSTEXPR_17 ParseFlags compound_parse_flag(const char ch) const noexcept {
                return quote_escape_flag(parse_flag(ch), this->quote_escape);
            }
 
            CONSTEXPR bool utf8_bom() const { return this->_utf8_bom; }
 
            void set_output(RowCollection& rows) { this->_records = &rows; }
 
        protected:
            CSVRow current_row;
            RawCSVDataPtr data_ptr = nullptr;
            ColNamesPtr _col_names = nullptr;
            CSVFieldList* fields = nullptr;
            int field_start = UNINITIALIZED_FIELD;
            size_t field_length = 0;
 
            ParseFlagMap _parse_flags;
 
            bool _eof = false;
 
            size_t source_size = 0;
 
            CONSTEXPR bool no_chunk() const { return this->source_size < ITERATION_CHUNK_SIZE; }
 
            size_t parse();
 
            void reset_data_ptr();
        private:
            WhitespaceMap _ws_flags;
            bool quote_escape = false;
            bool field_has_double_quote = false;
 
            size_t data_pos = 0;
 
            bool unicode_bom_scan = false;
            bool _utf8_bom = false;
 
            RowCollection* _records = nullptr;
 
            CONSTEXPR_17 bool ws_flag(const char ch) const noexcept {
                return _ws_flags.data()[ch + CHAR_OFFSET];
            }
 
            size_t& current_row_start() {
                return this->current_row.data_start;
            }
 
            void parse_field() noexcept;
 
            void push_field();
 
            void push_row();
 
            void trim_utf8_bom();
        };
 
        template<typename TStream,
            csv::enable_if_t<std::is_base_of<std::istream, TStream>::value, int>  = 0>
        std::string get_csv_head(TStream &source) {
            auto tellg = source.tellg();
            std::string head;
            std::getline(source, head);
            source.seekg(tellg);
            return head;
        }
 
        CSV_INLINE std::string get_csv_head(csv::string_view filename, size_t file_size);
 
        template<typename TStream>
        class StreamParser: public IBasicCSVParser {
            using RowCollection = ThreadSafeDeque<CSVRow>;
 
        public:
            StreamParser(TStream& source,
                const CSVFormat& format,
                const ColNamesPtr& col_names = nullptr
            ) : IBasicCSVParser(format, col_names), _source(source) {}
 
            StreamParser(
                TStream& source,
                internals::ParseFlagMap parse_flags,
                internals::WhitespaceMap ws_flags) :
                IBasicCSVParser(parse_flags, ws_flags),
                _source(source)
            {}
 
            ~StreamParser() {}
 
            void next(size_t bytes = ITERATION_CHUNK_SIZE) override {
                if (this->eof()) return;
 
                // Reset parser state
                this->field_start = UNINITIALIZED_FIELD;
                this->field_length = 0;
                this->reset_data_ptr();
                this->data_ptr->_data = std::make_shared<std::string>();
 
                if (source_size == 0) {
                    const auto start = _source.tellg();
                    _source.seekg(0, std::ios::end);
                    const auto end = _source.tellg();
                    _source.seekg(0, std::ios::beg);
 
                    source_size = end - start;
                }
 
                // Read data into buffer
                size_t length = std::min(source_size - stream_pos, bytes);
                std::unique_ptr<char[]> buff(new char[length]);
                _source.seekg(stream_pos, std::ios::beg);
                _source.read(buff.get(), length);
                stream_pos = _source.tellg();
                ((std::string*)(this->data_ptr->_data.get()))->assign(buff.get(), length);
 
                // Create string_view
                this->data_ptr->data = *((std::string*)this->data_ptr->_data.get());
 
                // Parse
                this->current_row = CSVRow(this->data_ptr);
                size_t remainder = this->parse();
 
                if (stream_pos == source_size || no_chunk()) {
                    this->_eof = true;
                    this->end_feed();
                }
                else {
                    this->stream_pos -= (length - remainder);
                }
            }
 
        private:
            TStream& _source;
            size_t stream_pos = 0;
        };
 
        class MmapParser : public IBasicCSVParser {
        public:
            MmapParser(csv::string_view filename,
                const CSVFormat& format,
                const ColNamesPtr& col_names = nullptr
            ) : IBasicCSVParser(format, col_names) {
                this->_filename = filename.data();
                this->source_size = get_file_size(filename);
            };
 
            ~MmapParser() {}
 
            void next(size_t bytes) override;
 
        private:
            std::string _filename;
            size_t mmap_pos = 0;
        };
    }
}