#pragma once

#include <ctime>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include "abstract_table_generator.hpp"
#include "benchmark_config.hpp"
#include "resolve_type.hpp"
#include "tpcc_random_generator.hpp"

namespace hyrise {

class Table;

using TPCCTableGeneratorFunctions = std::unordered_map<std::string, std::function<std::shared_ptr<Table>()>>;

class TPCCTableGenerator : public AbstractTableGenerator {
  // following TPC-C v5.11.0
 public:
  TPCCTableGenerator(size_t num_warehouses, const std::shared_ptr<BenchmarkConfig>& benchmark_config);

  // Convenience constructor for creating a TPCCTableGenerator without a benchmarking context.
  explicit TPCCTableGenerator(size_t num_warehouses, ChunkOffset chunk_size = Chunk::DEFAULT_SIZE);

  std::shared_ptr<Table> generate_item_table();

  std::shared_ptr<Table> generate_warehouse_table();

  std::shared_ptr<Table> generate_stock_table();

  std::shared_ptr<Table> generate_district_table();

  std::shared_ptr<Table> generate_customer_table();

  std::shared_ptr<Table> generate_history_table();

  using OrderLineCounts = std::vector<std::vector<std::vector<size_t>>>;

  OrderLineCounts generate_order_line_counts() const;

  std::shared_ptr<Table> generate_order_table(const OrderLineCounts& order_line_counts);

  std::shared_ptr<Table> generate_order_line_table(const OrderLineCounts& order_line_counts);

  std::shared_ptr<Table> generate_new_order_table();

  std::unordered_map<std::string, BenchmarkTableInfo> generate() override;

  const size_t _num_warehouses;
  const time_t _current_date = std::time(nullptr);

 protected:
  IndexesByTable _indexes_by_table() const override;
  void _add_constraints(std::unordered_map<std::string, BenchmarkTableInfo>& table_info_by_name) const override;

  template <typename T>
  std::vector<std::optional<T>> _generate_inner_order_line_column(
      const std::vector<size_t>& indices, OrderLineCounts order_line_counts,
      const std::function<std::optional<T>(const std::vector<size_t>&)>& generator_function);

  template <typename T>
  void _add_order_line_column(std::vector<Segments>& segments_by_chunk, TableColumnDefinitions& column_definitions,
                              std::string name, std::shared_ptr<std::vector<size_t>> cardinalities,
                              OrderLineCounts order_line_counts,
                              const std::function<std::optional<T>(const std::vector<size_t>&)>& generator_function);

  // Used to generate not only random numbers, but also non-uniform numbers and random last names as defined by the
  // TPC-C Specification.
  static thread_local TPCCRandomGenerator _random_gen;

  /**
   * In TPCC and TPCH table sizes are usually defined relatively to each other.
   * E.g. the specification defines that there are 10 districts for each warehouse.
   *
   * A trivial approach to implement this in our table generator would be to iterate in nested loops and add all rows.
   * However, this makes it hard to take care of a certain chunk size. With nested loops
   * chunks only contain as many rows as there are iterations in the most inner loop.
   *
   * In this method we basically generate the whole column in a single loop,
   * so that we can easily split when a Chunk is full. To do that we have all the cardinalities of the influencing
   * tables:
   * E.g. for the CUSTOMER table we have the following cardinalities:
   * indices[0] = warehouse_size = 1
   * indices[1] = district_size = 10
   * indices[2] = customer_size = 3000
   * So in total we have to generate 1*10*3000 = 30 000 customers.
   *
   * Note that this is not a general purpose function. The implementation of _add_column checks whether NULL values
   * were actually passed and sets the column's nullable flag accordingly. As such, if this method is used with an
   * generator that returns optionals but no nullopts, the column will not be nullable. For TPC-C, this is fine.
   *
   * @tparam T                  the type of the column
   * @param table               the column shall be added to this table as well as column metadata
   * @param name                the name of the column
   * @param cardinalities       the cardinalities of the different 'nested loops',
   *                            e.g. 10 districts per warehouse results in {1, 10}
   * @param generator_function  a lambda function to generate a vector of values for this column
   */
  template <typename T>
  void _add_column(std::vector<Segments>& segments_by_chunk, TableColumnDefinitions& column_definitions,
                   std::string name, std::shared_ptr<std::vector<size_t>> cardinalities,
                   const std::function<std::vector<std::optional<T>>(const std::vector<size_t>&)>& generator_function) {
    const auto chunk_size = _benchmark_config->chunk_size;

    auto is_first_column = column_definitions.size() == 0;

    auto has_null_value = false;

    /**
     * Calculate the total row count for this column based on the cardinalities of the influencing tables.
     * For the CUSTOMER table this calculates 1*10*3000
     */
    auto loop_count =
        std::accumulate(std::begin(*cardinalities), std::end(*cardinalities), 1u, std::multiplies<size_t>());

    auto data = pmr_vector<T>{};
    data.reserve(chunk_size);

    auto null_values = pmr_vector<bool>{};
    null_values.reserve(chunk_size);

    /**
     * The loop over all records that the final column of the table will contain, e.g. loop_count = 30 000 for CUSTOMER
     */
    auto row_index = size_t{0};

    for (auto loop_index = size_t{0}; loop_index < loop_count; ++loop_index) {
      auto indices = std::vector<size_t>(cardinalities->size());

      /**
       * Calculate indices for internal loops
       *
       * We have to take care of writing IDs for referenced table correctly, e.g. when they are used as foreign key.
       * In that case the 'generator_function' has to be able to access the current index of our loops correctly,
       * which we ensure by defining them here.
       *
       * For example for CUSTOMER:
       * WAREHOUSE_ID | DISTRICT_ID | CUSTOMER_ID
       * indices[0]   | indices[1]  | indices[2]
       */
      for (auto loop = size_t{0}; loop < cardinalities->size(); ++loop) {
        auto divisor = std::accumulate(std::begin(*cardinalities) + loop + 1, std::end(*cardinalities), 1u,
                                       std::multiplies<size_t>());
        indices[loop] = (loop_index / divisor) % cardinalities->at(loop);
      }

      /**
       * Actually generating and adding values.
       * Pass in the previously generated indices to use them in 'generator_function',
       * e.g. when generating IDs.
       * We generate a vector of values with variable length
       * and iterate it to add to the output segment.
       */
      auto values = generator_function(indices);
      for (const auto& value : values) {
        if (value) {
          data.emplace_back(std::move(*value));
        } else {
          data.emplace_back(T{});
          has_null_value = true;
        }
        null_values.emplace_back(!value);

        // write output chunks if segment size has reached chunk_size
        if (row_index % chunk_size == chunk_size - 1) {
          auto value_segment = has_null_value
                                   ? std::make_shared<ValueSegment<T>>(std::move(data), std::move(null_values))
                                   : std::make_shared<ValueSegment<T>>(std::move(data));

          // add Chunk if it is the first column, e.g. WAREHOUSE_ID in the example above
          if (is_first_column) {
            segments_by_chunk.emplace_back();
            segments_by_chunk.back().emplace_back(value_segment);
          } else {
            ChunkID chunk_id{static_cast<uint32_t>(row_index / chunk_size)};
            segments_by_chunk[chunk_id].emplace_back(value_segment);
          }

          // reset data
          data = {};
          data.reserve(chunk_size);

          null_values = {};
          null_values.reserve(chunk_size);
        }
        ++row_index;
      }
    }

    // write partially filled last chunk
    if (row_index % chunk_size != 0) {
      auto value_segment = has_null_value ? std::make_shared<ValueSegment<T>>(std::move(data), std::move(null_values))
                                          : std::make_shared<ValueSegment<T>>(std::move(data));

      if (is_first_column) {
        segments_by_chunk.emplace_back();
        segments_by_chunk.back().emplace_back(value_segment);
      } else {
        ChunkID chunk_id{static_cast<ChunkID::base_type>(row_index / chunk_size)};
        segments_by_chunk[chunk_id].emplace_back(value_segment);
      }
    }

    // add column definition
    auto data_type = data_type_from_type<T>();
    column_definitions.emplace_back(name, data_type, has_null_value);
  }

  /**
   * This method simplifies the interface for columns where only a single element is added in the inner loop.
   *
   * @tparam T                  the type of the column
   * @param table               the column shall be added to this table as well as column metadata
   * @param name                the name of the column
   * @param cardinalities       the cardinalities of the different 'nested loops',
   *                            e.g. 10 districts per warehouse results in {1, 10}
   * @param generator_function  a lambda function to generate a value for this column
   */
  template <typename T>
  void _add_column(std::vector<Segments>& segments_by_chunk, TableColumnDefinitions& column_definitions,
                   std::string name, std::shared_ptr<std::vector<size_t>> cardinalities,
                   const std::function<T(const std::vector<size_t>&)>& generator_function) {
    const std::function<std::vector<T>(const std::vector<size_t>&)> wrapped_generator_function =
        [generator_function](const std::vector<size_t>& indices) {
          return std::vector<T>({generator_function(indices)});
        };
    _add_column(segments_by_chunk, column_definitions, name, cardinalities, wrapped_generator_function);
  }

  /**
   * Similar to the previous _add_column version, this is a shortcut for columns where only a single element is added
   * in the inner loop. The difference is that this version accepts a lambda that produces an std::optional<T> (for
   * NULL values).
   *
   * Note that this is not a general purpose function. The implementation of _add_column checks whether NULL values
   * were actually passed and sets the column's nullable flag accordingly. As such, if this method is used with a
   * generator that returns optionals but no nullopts, the column will not be nullable. For TPC-C, this is fine.
   */
  template <typename T>
  void _add_column(std::vector<Segments>& segments_by_chunk, TableColumnDefinitions& column_definitions,
                   std::string name, std::shared_ptr<std::vector<size_t>> cardinalities,
                   const std::function<std::optional<T>(const std::vector<size_t>&)>& generator_function) {
    const std::function<std::vector<std::optional<T>>(const std::vector<size_t>&)> wrapped_generator_function =
        [generator_function](const std::vector<size_t>& indices) {
          return std::vector<std::optional<T>>({generator_function(indices)});
        };
    _add_column(segments_by_chunk, column_definitions, name, cardinalities, wrapped_generator_function);
  }
};
}  // namespace hyrise
