// Step 5 -- the payoff: a real query, end to end.
//
// Parsers are means. The task a user actually runs is something like
//
//     SELECT category, COUNT(*), SUM(value) FROM bench.csv GROUP BY category
//
// This step runs exactly that query two ways and prints both times:
//
//   * naive:  step 1's getline/copy parser + std::stod, the way a quick
//             script would do it;
//   * fast:   step 4's zero-copy scanner + std::from_chars straight off the
//             string_view bytes (no NUL-terminated copy, no locale), and a
//             fixed-size open-addressing table keyed by the category view --
//             no std::unordered_map, no per-node allocations, no string keys.
//
// Notes for the skeptical reader:
//   * std::from_chars<double> is verified working on this toolchain (Apple
//     clang 17) even though it does not advertise __cpp_lib_to_chars; the
//     build fails loudly if it is absent, it cannot silently misparse.
//   * The two pipelines must AGREE: same row order, same double accumulator,
//     so the sums must match bit for bit, and the group counts exactly.
//     They are also checked against gen_data.py's printed reference.
//
// Build:  c++ -std=c++20 -O2 step5_typed.cpp -o step5
#include "csv_test_common.hpp"
#include <charconv>
#include <fcntl.h>
#include <fstream>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <cstdint>
#include <cstring>
#include <string>
#include <string_view>
#include <vector>

// ---- fast pipeline: step 4's machinery (views + SWAR scan) --------------------
struct Mapped {
    const char* data = nullptr;
    size_t size = 0;
    Mapped(const char* path) {
        int fd = ::open(path, O_RDONLY);
        if (fd < 0) { std::perror(path); std::exit(1); }
        struct stat st{};
        ::fstat(fd, &st);
        size = size_t(st.st_size);
        data = (const char*)::mmap(nullptr, size, PROT_READ, MAP_PRIVATE, fd, 0);
        ::close(fd);
        if (data == MAP_FAILED) { std::perror("mmap"); std::exit(1); }
    }
    ~Mapped() { if (data) ::munmap((void*)data, size); }
};

struct Field { std::string_view raw; bool needs_unescape; };

static void unescape_into(std::string_view raw, std::string& out) {
    out.clear();
    for (size_t i = 0; i < raw.size(); ++i) {
        out += raw[i];
        if (raw[i] == '"') ++i;
    }
}

static constexpr uint64_t kOnes = 0x0101010101010101ull;
static constexpr uint64_t kHighs = 0x8080808080808080ull;
static inline uint64_t match_byte(uint64_t x, uint8_t b) {
    uint64_t v = x ^ (kOnes * b);
    return (v - kOnes) & ~v & kHighs;
}
static inline const char* scan_unquoted(const char* p, const char* end) {
    while (end - p >= 8) {
        uint64_t x;
        std::memcpy(&x, p, 8);
        uint64_t m = match_byte(x, ',') | match_byte(x, '\r') | match_byte(x, '\n');
        if (m) return p + (__builtin_ctzll(m) >> 3);
        p += 8;
    }
    while (p < end && *p != ',' && *p != '\r' && *p != '\n') ++p;
    return p;
}
static inline const char* scan_quoted(const char* p, const char* end) {
    while (end - p >= 8) {
        uint64_t x;
        std::memcpy(&x, p, 8);
        if (uint64_t m = match_byte(x, '"')) return p + (__builtin_ctzll(m) >> 3);
        p += 8;
    }
    while (p < end && *p != '"') ++p;
    return p;
}

template <typename OnRecord>
static void parse_fast(const char* p, const char* end, OnRecord on_record) {
    std::vector<Field> rec;
    rec.reserve(16);
    while (p < end) {
        rec.clear();
        for (;;) {
            const char* begin;
            if (p < end && *p == '"') {
                begin = ++p;
                bool esc = false;
                for (;;) {
                    p = scan_quoted(p, end);
                    if (p + 1 < end && p[1] == '"') { esc = true; p += 2; }
                    else break;
                }
                rec.push_back({{begin, size_t(p - begin)}, esc});
                if (p < end) ++p;
            } else {
                begin = p;
                p = scan_unquoted(p, end);
                rec.push_back({{begin, size_t(p - begin)}, false});
            }
            if (p >= end) break;
            char c = *p++;
            if (c == ',') continue;
            if (c == '\r' && p < end && *p == '\n') ++p;
            break;
        }
        on_record(rec);
    }
}

// ---- the group-by table --------------------------------------------------------
// Categories are a handful of distinct strings, so a 64-slot linear-probe
// table over string_view keys does what unordered_map does with zero
// allocations and no hashing library. FNV-1a is plenty.
struct GroupTable {
    static constexpr size_t N = 64;                   // power of two, > #groups
    std::string_view keys[N]{};
    uint64_t counts[N]{};
    static uint64_t hash(std::string_view s) {
        uint64_t h = 1469598103934665603ull;
        for (char c : s) { h ^= (unsigned char)c; h *= 1099511628211ull; }
        return h;
    }
    void add(std::string_view k) {
        size_t i = hash(k) & (N - 1);
        for (;;) {
            if (counts[i] == 0) { keys[i] = k; counts[i] = 1; return; }
            if (keys[i] == k) { ++counts[i]; return; }
            i = (i + 1) & (N - 1);
        }
    }
};

// ---- naive pipeline: step 1's parser, verbatim ---------------------------------
static std::vector<std::string> split_record(const std::string& rec) {
    std::vector<std::string> out;
    std::string cur;
    bool in_quotes = false;
    for (size_t i = 0; i < rec.size(); ++i) {
        char c = rec[i];
        if (in_quotes) {
            if (c == '"') {
                if (i + 1 < rec.size() && rec[i + 1] == '"') { cur += '"'; ++i; }
                else in_quotes = false;
            } else cur += c;
        } else {
            if (c == '"') in_quotes = true;
            else if (c == ',') { out.push_back(std::move(cur)); cur.clear(); }
            else cur += c;
        }
    }
    out.push_back(std::move(cur));
    return out;
}
static bool quotes_balanced(const std::string& s) {
    size_t n = 0;
    for (char c : s) n += (c == '"');
    return n % 2 == 0;
}

struct QueryResult { double sum = 0; uint64_t rows = 0; GroupTable groups; };

static QueryResult query_naive(const char* path) {
    QueryResult q;
    std::ifstream f(path, std::ios::binary);
    std::string line, record;
    // string keys must outlive the table's views: intern them here
    static const std::string cats[] = {"alpha", "beta", "gamma", "delta",
                                       "epsilon", "widgets, large", "zeta", "eta"};
    bool header = true;
    while (std::getline(f, line)) {
        record = line;
        while (!quotes_balanced(record) && std::getline(f, line)) {
            record += '\n';
            record += line;
        }
        if (!record.empty() && record.back() == '\r') record.pop_back();
        auto fields = split_record(record);
        if (header) { header = false; continue; }
        if (fields.size() < 4) continue;
        q.sum += std::stod(fields[3]);
        ++q.rows;
        for (const auto& c : cats)
            if (fields[1] == c) { q.groups.add(c); break; }
    }
    return q;
}

static QueryResult query_fast(const Mapped& m) {
    QueryResult q;
    std::string scratch;
    bool header = true;
    parse_fast(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
        if (header) { header = false; return; }
        if (rec.size() < 4) return;
        double v = 0;
        auto sv = rec[3].raw;
        std::from_chars(sv.data(), sv.data() + sv.size(), v);
        q.sum += v;
        ++q.rows;
        // category views point into the mapping, which outlives the table --
        // except unescaped ones ("widgets, large" is quoted but has no ""),
        // so raw IS the cooked value for every category in this data.
        q.groups.add(rec[1].raw);
    });
    return q;
}

int main() {
    // correctness gate first, same as every step (fast pipeline parses edge.csv)
    auto parse_all = [](const char* path) {
        Mapped m(path);
        std::vector<std::vector<std::string>> rows;
        std::string scratch;
        parse_fast(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
            std::vector<std::string> out;
            for (const Field& f : rec) {
                if (f.needs_unescape) { unescape_into(f.raw, scratch); out.push_back(scratch); }
                else out.emplace_back(f.raw);
            }
            rows.push_back(std::move(out));
        });
        return rows;
    };
    if (!run_edge_checks("edge.csv", parse_all)) return 1;

    Mapped m("bench.csv");
    const double mb = m.size / 1e6;

    double t0 = now_secs();
    QueryResult naive = query_naive("bench.csv");
    double t_naive = now_secs() - t0;

    double best_fast = 1e30;
    QueryResult fast;
    for (int rep = 0; rep < 3; ++rep) {
        t0 = now_secs();
        fast = query_fast(m);
        double dt = now_secs() - t0;
        if (dt < best_fast) best_fast = dt;
    }

    std::printf("query: SELECT category, COUNT(*), SUM(value) GROUP BY category\n");
    std::printf("  naive (step-1 parser + stod):        %.3f s  (%.0f MB/s)\n", t_naive, mb / t_naive);
    std::printf("  fast  (views + SWAR + from_chars):   %.3f s  (%.0f MB/s)  -> %.1fx\n",
                best_fast, mb / best_fast, t_naive / best_fast);
    std::printf("  sum(value): naive %.6f  fast %.6f  %s\n", naive.sum, fast.sum,
                naive.sum == fast.sum ? "BITWISE EQUAL" : "MISMATCH");
    bool groups_ok = true;
    for (size_t i = 0; i < GroupTable::N; ++i) {
        if (fast.groups.counts[i] == 0) continue;
        // find same key in naive table
        uint64_t want = 0;
        for (size_t j = 0; j < GroupTable::N; ++j)
            if (naive.groups.keys[j] == fast.groups.keys[i]) want = naive.groups.counts[j];
        if (fast.groups.counts[i] != want) groups_ok = false;
        std::printf("  count[%.*s] = %llu\n", (int)fast.groups.keys[i].size(),
                    fast.groups.keys[i].data(), (unsigned long long)fast.groups.counts[i]);
    }
    std::printf("  group counts: %s\n", groups_ok ? "MATCH" : "MISMATCH");
    if (naive.sum != fast.sum || !groups_ok) return 1;

    std::printf("BENCH step=step5_typed secs=%.4f rows=%llu fields=0 mbps=%.0f allocs=0 bytesum=0\n",
                best_fast, (unsigned long long)fast.rows, mb / best_fast);
    return 0;
}