#pragma once

#include <memory>
#include <string>

#ifdef __AVX512VL__
#include <x86intrin.h>
#endif

#include <array>
#include <atomic>

#include "operators/operator_performance_data.hpp"
#include "storage/pos_lists/row_id_pos_list.hpp"
#include "storage/segment_iterables.hpp"
#include "storage/segment_iterables/any_segment_iterator.hpp"
#include "types.hpp"
#include "utils/performance_warning.hpp"

namespace hyrise {

/**
 * @brief the base class of all table scan impls
 */
class AbstractTableScanImpl {
 public:
  virtual ~AbstractTableScanImpl() = default;

  virtual std::string description() const = 0;

  virtual std::shared_ptr<RowIDPosList> scan_chunk(ChunkID chunk_id) = 0;

  std::atomic_size_t num_chunks_with_early_out{0};
  std::atomic_size_t num_chunks_with_all_rows_matching{0};
  std::atomic_size_t num_chunks_with_binary_search{0};

 protected:
  /**
   * @defgroup The hot loop of the table scan
   * @{
   */

  template <bool CheckForNull, typename BinaryFunctor, typename LeftIterator>
  static void _scan_with_iterators(const BinaryFunctor func, LeftIterator left_it, const LeftIterator left_end,
                                   const ChunkID chunk_id, RowIDPosList& matches_out) {
    // Can't use a default argument for this because default arguments are non-type deduced contexts
    auto false_type = std::false_type{};
    _scan_with_iterators<CheckForNull>(func, left_it, left_end, chunk_id, matches_out, false_type);
  }

  template <bool CheckForNull, typename BinaryFunctor, typename LeftIterator, typename RightIterator>
  // This is a function that is critical for our performance. We want the compiler to try its best in optimizing it.
  // Also, we want all functions called inside to be inlined (flattened). GCC seems to like for this to not be inlined
  // itself.
  static void __attribute__((hot, flatten, noinline))
  _scan_with_iterators(const BinaryFunctor func, LeftIterator left_it, const LeftIterator left_end,
                       const ChunkID chunk_id, RowIDPosList& matches_out, [[maybe_unused]] RightIterator right_it) {
    // The major part of the table is scanned using SIMD. Only the remainder is handled in this method.
    // For a description of the SIMD code, have a look at the comments in that method.
    // To reduce compile time, SIMD scanning is not used for for ColumnVsColumnScans. Also, string comparisons are more
    // expensive than the scan itself, so we disable SIMD for these, too.
    if constexpr (std::is_same_v<RightIterator, std::false_type> &&
                  !std::is_same_v<std::decay_t<decltype(left_it->value())>, pmr_string>) {
      _simd_scan_with_iterators<CheckForNull>(func, left_it, left_end, chunk_id, matches_out, right_it);
    }

    // Do the remainder the easy way. If we did not use the SIMD optimization above, left_it was not yet touched, so we
    // iterate over the entire input data.
    for (; left_it != left_end; ++left_it) {
      const auto left = *left_it;
      if constexpr (std::is_same_v<RightIterator, std::false_type>) {
        if ((!CheckForNull || !left.is_null()) && func(left)) {
          matches_out.emplace_back(chunk_id, left.chunk_offset());
        }
      } else {
        const auto right = *right_it;
        if ((!CheckForNull || (!left.is_null() && !right.is_null())) && func(left, right)) {
          matches_out.emplace_back(chunk_id, left.chunk_offset());
        }
        ++right_it;
      }
    }
  }

  template <bool CheckForNull, typename BinaryFunctor, typename LeftIterator, typename RightIterator>
  static void _simd_scan_with_iterators(const BinaryFunctor func, LeftIterator& left_it, const LeftIterator left_end,
                                        const ChunkID chunk_id, RowIDPosList& matches_out,
                                        [[maybe_unused]] RightIterator& right_it) {
    // Concept: Partition the vector into blocks of BLOCK_SIZE entries. The remainder is handled by the caller without
    // optimization. We first check if the rows match and set the `mask` to 1 at the appropriate positions.
    // If the mask is empty, we continue with the next block. If at least one row matches (and the mask is not empty),
    // we fill `offsets` with the ChunkOffsets of ALL rows in the block, not only those that match. A SIMD optimization
    // is used to move the matching rows into `matches_out`. There, we do not push_back/emplace_back values, but
    // resize the vector first and directly write values into the next position, given by matches_out_index. This
    // avoids calls into the stdlib from the hot loop.
    //
    // Most of the performance gain does not come from scanning the input data but from inserting into matches_out more
    // efficiently. That is why you will see bigger benefits for scans that select more rows.
    //
    // Finally, this implementation is only better if we are on a machine that supports AVX-512 VL. If we are not, we
    // still do one iteration, simply to reduce the amount of dead / diverging code. After that, we return to the
    // non-SIMD scan.

    // We assume a maximum SIMD register size of 256 bit. Smaller registers simply lead to the inner loop being unrolled
    // more than once. Even on machines with 512-bit SIMD registers, we prefer to use 256 bits, because current Intel
    // CPUs clock down when 512-bit is used. If you want to run this with 512-bit registers, you need to change the
    // SIMD_SIZE here as well as replace _m256 with _m512 twice below.
    constexpr size_t SIMD_SIZE = 256 / 8;
    constexpr size_t BLOCK_SIZE = SIMD_SIZE / sizeof(ChunkOffset);

    // The index at which we will write the next matching row
    auto matches_out_index = matches_out.size();

    // Make sure that we have enough space for the first iteration. We might resize later on.
    matches_out.resize(matches_out.size() + BLOCK_SIZE, RowID{chunk_id, ChunkOffset{0}});

    // As we access the offsets after we already moved the iterator, we need a copy of it. Creating this copy outside
    // of the while loop keeps the surprisingly high costs for copying an iterator to a minimum.
    auto left_it_for_offsets = left_it;

    // Continue the following until we have too few rows left to run over a whole block
    while (static_cast<size_t>(left_end - left_it) > BLOCK_SIZE) {
#ifdef __AVX512VL__
      // __mmask8 would be sufficient, but either GCC or the CPU likes a 16-bit mask better, which is good for the
      // performance.
      __mmask16 mask = 0;
#else
      // Using uint16_t here for consistency with the above.
      uint16_t mask = 0;
#endif

      // The OpenMP Pragma makes the compiler try harder to vectorize this and gives some hints to help with this.
      // We do not use the OpenMP runtime, but only the compiler pragmas (look up -fopenmp-simd).

      // This empty block is used to convince clang-format to keep the pragma indented
      // NOLINTNEXTLINE
      {}  // clang-format off
      #pragma omp simd reduction(|:mask) safelen(BLOCK_SIZE)
      // clang-format on
      for (auto index = size_t{0}; index < BLOCK_SIZE; ++index) {
        // Fill `mask` with 1s at positions where the condition is fulfilled
        const auto& left = *left_it;

        if constexpr (std::is_same_v<RightIterator, std::false_type>) {
          mask |= ((!CheckForNull | !left.is_null()) && func(left)) << index;
        } else {
          const auto& right = *right_it;
          mask |= ((!CheckForNull | (!left.is_null() && !right.is_null())) & func(left, right)) << index;
        }

        ++left_it;
        if constexpr (!std::is_same_v<RightIterator, std::false_type>) {
          ++right_it;
        }
      }

      if (!mask) {
        left_it_for_offsets += BLOCK_SIZE;
        continue;
      }

      // Next, write *all* offsets in the block into `offsets`
      auto offsets = std::array<ChunkOffset, BLOCK_SIZE>{};

      if constexpr (!std::is_base_of_v<AbstractPointAccessSegmentIterator<std::decay_t<decltype(left_it)>,
                                                                          std::decay_t<decltype(*left_it)>,
                                                                          std::decay_t<decltype(left_it)>>,
                                       std::decay_t<decltype(left_it)>>) {
        // Fast path: If this is a sequential iterator, we know that the chunk offsets are incremented by 1, so we can
        // save us the memory lookup
        const auto first_offset = left_it_for_offsets->chunk_offset();

        // NOLINTNEXTLINE
        {}  // clang-format off
        #pragma omp simd safelen(BLOCK_SIZE)
        // clang-format on
        for (auto index = size_t{0}; index < BLOCK_SIZE; ++index) {
          offsets[index] = first_offset + static_cast<ChunkOffset>(index);
        }

        left_it_for_offsets += BLOCK_SIZE;
      } else {
        // Slow path - the chunk offsets are not guaranteed to be linear

        // NOLINTNEXTLINE
        {}  // clang-format off
        #pragma omp simd safelen(BLOCK_SIZE)
        // clang-format on
        for (auto index = size_t{0}; index < BLOCK_SIZE; ++index) {
          offsets[index] = left_it_for_offsets->chunk_offset();
          ++left_it_for_offsets;
        }
      }

      // Now write the matches into matches_out.
#ifndef __AVX512VL__
      // "Slow" path for non-AVX512VL systems
      for (auto index = size_t{0}; index < BLOCK_SIZE; ++index) {
        if (mask >> index & 1) {
          matches_out[matches_out_index++].chunk_offset = offsets[index];
        }
      }

      // We have done one iteration. As described above, we stop the SIMD code here, because it won't be faster but
      // might be significantly slower.
      break;
#else
      // Fast path for AVX512VL systems

      // Compress `offsets`, i.e., move all values where the mask is set to 1 to the front
      auto offsets_simd =
          _mm256_maskz_compress_epi32(static_cast<unsigned char>(mask), reinterpret_cast<__m256i&>(offsets));

      // Copy all offsets into `matches_out` - even those that are set to 0 (which are located at the end). This does
      // not matter because they will be overwritten in the next round anyway. Copying more than necessary is better
      // than stopping at the number of matching rows because we do not need a branch for this. The loop will be
      // vectorized automatically.

      // NOLINTNEXTLINE
      {}  // clang-format off
      #pragma omp simd safelen(BLOCK_SIZE)
      // clang-format on
      for (auto index = size_t{0}; index < BLOCK_SIZE; ++index) {
        matches_out[matches_out_index + index].chunk_offset = (reinterpret_cast<ChunkOffset*>(&offsets_simd))[index];
      }

      // Count the number of matches and increase the index of the next write to matches_out accordingly
      matches_out_index += __builtin_popcount(mask);
#endif

      // As we write directly into the matches_out vector, we have to make sure that is big enough. We grow the vector
      // more aggressively than its default behavior as the potentially wasted space is only ephemeral.
      if (matches_out_index + BLOCK_SIZE >= matches_out.size()) {
        matches_out.resize((BLOCK_SIZE + matches_out.size()) * 3, RowID{chunk_id, ChunkOffset{0}});
      }
    }

    // Remove all entries that we have overallocated
    matches_out.resize(matches_out_index);

    // The remainder is now done by the regular scan
  }

  /**@}*/
};

}  // namespace hyrise
