// Arrow/Parquet reader for FinData bars (schema v1). Column chunks are read with the Arrow C++ API; numeric columns
// are copied once into the BarSeries vectors (a true zero-copy view is only possible for a single contiguous chunk of
// float64 without nulls — the reader records whether that was the case in `metadata["zero_copy_possible"]`).
#include "raid/data/io.hpp"
#include "raid/core/error.hpp"
#include <arrow/api.h>
#include <arrow/io/api.h>
#include <arrow/ipc/api.h>
#include <parquet/arrow/reader.h>
#include <unordered_set>

namespace raid::data {

namespace {
ParquetBars table_to_bars(const std::shared_ptr<arrow::Table>& table, const std::vector<std::string>& symbols, Interval iv) {
  ParquetBars out;
  if (table->schema()->metadata()) for (int i = 0; i < table->schema()->metadata()->size(); ++i) out.metadata[table->schema()->metadata()->key(i)] = table->schema()->metadata()->value(i);
  auto col = [&](const char* name) { auto c = table->GetColumnByName(name); RAID_REQUIRE(c != nullptr, io, std::string("missing column ") + name); return c; };
  auto ts = col("timestamp_utc"); auto sym = col("symbol_as_reported"); auto iid = table->GetColumnByName("instrument_id");   // may be absent (projected read)
  auto o = col("open"), h = col("high"), l = col("low"), c = col("close"), v = col("volume");
  std::unordered_set<std::string> keep(symbols.begin(), symbols.end());
  bool contiguous = (c->num_chunks() == 1) && (c->null_count() == 0);
  out.metadata["zero_copy_possible"] = contiguous ? "true" : "false";
  // Fast path: symbols are stored in runs (one symbol = one contiguous block in FinData exports); avoid a map lookup per row.
  BarSeries* cur = nullptr; std::string cur_sym;
  for (int ch = 0; ch < ts->num_chunks(); ++ch) {
    auto ta = std::static_pointer_cast<arrow::TimestampArray>(ts->chunk(ch));
    auto sa = std::static_pointer_cast<arrow::StringArray>(sym->chunk(ch));
    std::shared_ptr<arrow::StringArray> ia = iid ? std::static_pointer_cast<arrow::StringArray>(iid->chunk(ch)) : nullptr;
    auto oa = std::static_pointer_cast<arrow::DoubleArray>(o->chunk(ch)), ha = std::static_pointer_cast<arrow::DoubleArray>(h->chunk(ch));
    auto la = std::static_pointer_cast<arrow::DoubleArray>(l->chunk(ch)), ca = std::static_pointer_cast<arrow::DoubleArray>(c->chunk(ch));
    auto va = std::static_pointer_cast<arrow::DoubleArray>(v->chunk(ch));
    const auto unit = std::static_pointer_cast<arrow::TimestampType>(ta->type())->unit();
    i64 mult = 1; if (unit == arrow::TimeUnit::SECOND) mult = 1'000'000; else if (unit == arrow::TimeUnit::MILLI) mult = 1000; else if (unit == arrow::TimeUnit::NANO) mult = -1000;
    for (int64_t i = 0; i < ta->length(); ++i) {
      const std::string_view sv = sa->GetView(i);
      if (!cur || sv != cur_sym) { cur_sym = std::string(sv); if (!keep.empty() && !keep.count(cur_sym)) { cur = nullptr; continue; } cur = &out.series[cur_sym]; if (cur->symbol.empty()) { cur->symbol = cur_sym; cur->instrument_id = ia ? ia->GetString(i) : ("firstrate:" + std::string(std::static_pointer_cast<arrow::StringArray>(col("asset_class")->chunk(ch))->GetView(i)) + ":" + cur_sym); cur->interval = iv; cur->adjustment = "raw"; cur->reserve(4096); } }
      else if (!cur) continue;
      const i64 raw = ta->Value(i); const i64 us = mult > 0 ? raw * mult : raw / 1000;
      cur->push_back(Timestamp(us), oa->Value(i), ha->Value(i), la->Value(i), ca->Value(i), va->Value(i));
      ++out.rows_read;
    }
  }
  for (auto& [k, bs] : out.series) { auto st = bs.normalize(); out.duplicates_dropped += st.duplicates_removed; }
  return out;
}
}  // namespace

ParquetBars read_bars_parquet(const std::string& path, const std::vector<std::string>& symbols, Interval iv) {
  auto infile = arrow::io::ReadableFile::Open(path); RAID_REQUIRE(infile.ok(), io, "cannot open " + path + ": " + infile.status().ToString());
  auto reader_r = parquet::arrow::OpenFile(*infile, arrow::default_memory_pool()); RAID_REQUIRE(reader_r.ok(), io, "parquet open: " + reader_r.status().ToString());
  std::unique_ptr<parquet::arrow::FileReader> reader = std::move(*reader_r);
  // Column projection: only the 8 columns the estimators need (of 20) are decoded — the key I/O optimisation of RQ3.
  std::shared_ptr<arrow::Schema> schema; auto sst = reader->GetSchema(&schema); RAID_REQUIRE(sst.ok(), io, "parquet schema: " + sst.ToString());
  // (instrument_id is derived from symbol_as_reported for FinData exports: "firstrate:<class>:<SYMBOL>"; reading the string column costs more than every numeric column together)
  std::vector<int> idx; for (const char* name : {"symbol_as_reported", "asset_class", "timestamp_utc", "open", "high", "low", "close", "volume"}) { const int k = schema->GetFieldIndex(name); RAID_REQUIRE(k >= 0, io, std::string("missing column ") + name); idx.push_back(k); }
  reader->set_use_threads(true);
  std::shared_ptr<arrow::Table> table; auto st = reader->ReadTable(idx, &table); RAID_REQUIRE(st.ok(), io, "parquet read: " + st.ToString());
  return table_to_bars(table, symbols, iv);
}

ParquetBars read_bars_arrow_ipc(const std::string& path, Interval iv) {
  auto infile = arrow::io::ReadableFile::Open(path); RAID_REQUIRE(infile.ok(), io, "cannot open " + path);
  auto reader = arrow::ipc::RecordBatchStreamReader::Open(*infile); RAID_REQUIRE(reader.ok(), io, "arrow ipc open: " + reader.status().ToString());
  auto table = (*reader)->ToTable(); RAID_REQUIRE(table.ok(), io, "arrow ipc read: " + table.status().ToString());
  return table_to_bars(*table, {}, iv);
}

}  // namespace raid::data