basic_csv_parser.hpp

Go to the documentation of this file.

 
#pragma once
#include <algorithm>
#include <array>
#include <condition_variable>
#include <deque>
#include <fstream>
#include <memory>
#include <mutex>
#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"
 
namespace csv {
    namespace internals {
        HEDLEY_CONST CONSTEXPR_17 ParseFlagMap make_parse_flags(char delimiter) {
            std::array<ParseFlags, 256> ret = {};
            for (int i = -128; i < 128; i++) {
                const int arr_idx = i + 128;
                char ch = char(i);
 
                if (ch == delimiter)
                    ret[arr_idx] = ParseFlags::DELIMITER;
                else if (ch == '\r' || ch == '\n')
                    ret[arr_idx] = ParseFlags::NEWLINE;
                else
                    ret[arr_idx] = ParseFlags::NOT_SPECIAL;
            }
 
            return ret;
        }
 
        HEDLEY_CONST CONSTEXPR_17 ParseFlagMap make_parse_flags(char delimiter, char quote_char) {
            std::array<ParseFlags, 256> ret = make_parse_flags(delimiter);
            ret[(size_t)quote_char + 128] = ParseFlags::QUOTE;
            return ret;
        }
 
        HEDLEY_CONST CONSTEXPR_17 WhitespaceMap make_ws_flags(const char* ws_chars, size_t n_chars) {
            std::array<bool, 256> ret = {};
            for (int i = -128; i < 128; i++) {
                const int arr_idx = i + 128;
                char ch = char(i);
                ret[arr_idx] = false;
 
                for (size_t j = 0; j < n_chars; j++) {
                    if (ws_chars[j] == ch) {
                        ret[arr_idx] = 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);
 
        CSV_INLINE std::string get_csv_head(csv::string_view filename, size_t file_size);
 
        template<typename T>
        class ThreadSafeDeque {
        public:
            ThreadSafeDeque(size_t notify_size = 100) : _notify_size(notify_size) {};
            ThreadSafeDeque(const ThreadSafeDeque& other) {
                this->data = other.data;
                this->_notify_size = other._notify_size;
            }
 
            ThreadSafeDeque(const std::deque<T>& source) : ThreadSafeDeque() {
                this->data = source;
            }
 
            void clear() noexcept { this->data.clear(); }
 
            bool empty() const noexcept {
                return this->data.empty();
            }
 
            T& front() noexcept {
                return this->data.front();
            }
 
            T& operator[](size_t n) {
                return this->data[n];
            }
 
            void push_back(T&& item) {
                std::lock_guard<std::mutex> lock{ this->_lock };
                this->data.push_back(std::move(item));
 
                if (this->size() >= _notify_size) {
                    this->_cond.notify_all();
                }
            }
 
            T pop_front() noexcept {
                std::lock_guard<std::mutex> lock{ this->_lock };
                T item = std::move(data.front());
                data.pop_front();
                return item;
            }
 
            size_t size() const noexcept { return this->data.size(); }
 
            constexpr bool is_waitable() const noexcept { return this->_is_waitable; }
 
            void wait() {
                if (!is_waitable()) {
                    return;
                }
 
                std::unique_lock<std::mutex> lock{ this->_lock };
                this->_cond.wait(lock, [this] { return this->size() >= _notify_size || !this->is_waitable(); });
                lock.unlock();
            }
 
            typename std::deque<T>::iterator begin() noexcept {
                return this->data.begin();
            }
 
            typename std::deque<T>::iterator end() noexcept {
                return this->data.end();
            }
 
            void notify_all() {
                std::unique_lock<std::mutex> lock{ this->_lock };
                this->_is_waitable = true;
                this->_cond.notify_all();
            }
 
            void kill_all() {
                std::unique_lock<std::mutex> lock{ this->_lock };
                this->_is_waitable = false;
                this->_cond.notify_all();
            }
 
        private:
            bool _is_waitable = false;
            size_t _notify_size;
            std::mutex _lock;
            std::condition_variable _cond;
            std::deque<T> data;
        };
 
        constexpr const int UNINITIALIZED_FIELD = -1;
    }
 
    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 + 128];
            }
 
            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 + 128];
            }
 
            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>
        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(std::move(source)) {};
 
            StreamParser(
                TStream& source,
                internals::ParseFlagMap parse_flags,
                internals::WhitespaceMap ws_flags) :
                IBasicCSVParser(parse_flags, ws_flags),
                _source(std::move(source))
            {};
 
            ~StreamParser() {}
 
            void next(size_t bytes = ITERATION_CHUNK_SIZE) override {
                if (this->eof()) return;
 
                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;
        };
    }
}