<commit_before>#pragma once

#include <functional>
#include <memory>
#include <string>
#include <vector>

#include "abstract_operator.hpp"
#include "storage/reference_column.hpp"
#include "types.hpp"

namespace opossum {

template <typename T>
class TableScanImpl;

class TableScan : public AbstractOperator {
 public:
  TableScan(const std::shared_ptr<AbstractOperator> in, const std::string &filter_column_name, const std::string &op,
            const AllTypeVariant value);
  virtual void execute();
  virtual std::shared_ptr<Table> get_output() const;

 protected:
  virtual const std::string get_name() const;
  virtual uint8_t get_num_in_tables() const;
  virtual uint8_t get_num_out_tables() const;

  const std::unique_ptr<AbstractOperatorImpl> _impl;
};

template <typename T>
class TableScanImpl : public AbstractOperatorImpl {
 public:
  TableScanImpl(const std::shared_ptr<AbstractOperator> in, const std::string &filter_column_name,
                const std::string &op, const AllTypeVariant value)
      : _filter_value(type_cast<T>(value)),
        _table(in->get_output()),
        _filter_column_id(_table->get_column_id_by_name(filter_column_name)),
        _op(op),
        _output(new Table),
        _pos_list(new PosList) {
    for (size_t column_id = 0; column_id < _table->col_count(); ++column_id) {
      std::shared_ptr<ReferenceColumn> ref;
      if (auto ref_col = std::dynamic_pointer_cast<ReferenceColumn>(_table->get_chunk(0).get_column(column_id))) {
        ref = std::make_shared<ReferenceColumn>(ref_col->get_referenced_table(), column_id, _pos_list);
      } else {
        ref = std::make_shared<ReferenceColumn>(_table, column_id, _pos_list);
      }
      _output->add_column(_table->get_column_name(column_id), _table->get_column_type(column_id), false);
      _output->get_chunk(0).add_column(ref);
      // TODO(Anyone): do we want to distinguish between chunk tables and "reference tables"?
    }
  }

  template <typename Comp>
  void execute_with_operator() {
    Comp comp;
    if (auto ref_col = std::dynamic_pointer_cast<ReferenceColumn>(_table->get_chunk(0).get_column(_filter_column_id))) {
      auto val_table = ref_col->get_referenced_table();
      std::vector<std::vector<T>> values = {};
      for (size_t chunk = 0; chunk < val_table->chunk_count(); chunk++) {
        values.emplace_back(
            std::dynamic_pointer_cast<ValueColumn<T>>(val_table->get_chunk(chunk).get_column(_filter_column_id))
                ->get_values());
      }
      auto in_pos_list = ref_col->get_pos_list();

      for (size_t pos = 0; pos < in_pos_list->size(); pos++) {
        auto chunk_id = get_chunk_id_from_row_id((*in_pos_list)[pos]);
        auto chunk_offset = get_chunk_offset_from_row_id((*in_pos_list)[pos]);
        if (comp(values[chunk_id][chunk_offset], _filter_value)) {
          _pos_list->emplace_back(get_row_id_from_chunk_id_and_chunk_offset(chunk_id, chunk_offset));
        }
      }
    } else {
      for (ChunkID chunk_id = 0; chunk_id < _table->chunk_count(); ++chunk_id) {
        auto &chunk = _table->get_chunk(chunk_id);
        auto base_column = chunk.get_column(_filter_column_id);
        auto &values = std::dynamic_pointer_cast<ValueColumn<T>>(base_column)->get_values();
        for (ChunkOffset chunk_offset = 0; chunk_offset < chunk.size(); ++chunk_offset) {
          if (comp(values[chunk_offset], _filter_value)) {
            _pos_list->emplace_back(get_row_id_from_chunk_id_and_chunk_offset(chunk_id, chunk_offset));
          }
        }
      }
    }
  }

  void execute() {
    // Definining all possible operators here might appear odd. Chances are, however, that we will not
    // have a similar comparison anywhere else. Index scans, for example, would not use an adaptable binary
    // predicate, but will have to use different methods (lower_range, upper_range, ...) based on the
    // chosen operator. For now, we can save us some dark template magic by using the switch below.
    // DO NOT copy this code, however, without discussing if there is a better way to avoid code duplication.
<commit_msg>ensuring referenced columns during table scan are value columns<commit_after>#pragma once

#include <functional>
#include <memory>
#include <string>
#include <vector>

#include "abstract_operator.hpp"
#include "storage/reference_column.hpp"
#include "types.hpp"

namespace opossum {

template <typename T>
class TableScanImpl;

class TableScan : public AbstractOperator {
 public:
  TableScan(const std::shared_ptr<AbstractOperator> in, const std::string &filter_column_name, const std::string &op,
            const AllTypeVariant value);
  virtual void execute();
  virtual std::shared_ptr<Table> get_output() const;

 protected:
  virtual const std::string get_name() const;
  virtual uint8_t get_num_in_tables() const;
  virtual uint8_t get_num_out_tables() const;

  const std::unique_ptr<AbstractOperatorImpl> _impl;
};

template <typename T>
class TableScanImpl : public AbstractOperatorImpl {
 public:
  TableScanImpl(const std::shared_ptr<AbstractOperator> in, const std::string &filter_column_name,
                const std::string &op, const AllTypeVariant value)
      : _filter_value(type_cast<T>(value)),
        _table(in->get_output()),
        _filter_column_id(_table->get_column_id_by_name(filter_column_name)),
        _op(op),
        _output(new Table),
        _pos_list(new PosList) {
    for (size_t column_id = 0; column_id < _table->col_count(); ++column_id) {
      std::shared_ptr<ReferenceColumn> ref;
      if (auto ref_col = std::dynamic_pointer_cast<ReferenceColumn>(_table->get_chunk(0).get_column(column_id))) {
        ref = std::make_shared<ReferenceColumn>(ref_col->get_referenced_table(), column_id, _pos_list);
      } else {
        ref = std::make_shared<ReferenceColumn>(_table, column_id, _pos_list);
      }
      _output->add_column(_table->get_column_name(column_id), _table->get_column_type(column_id), false);
      _output->get_chunk(0).add_column(ref);
      // TODO(Anyone): do we want to distinguish between chunk tables and "reference tables"?
    }
  }

  template <typename Comp>
  void execute_with_operator() {
    Comp comp;
    if (auto ref_col = std::dynamic_pointer_cast<ReferenceColumn>(_table->get_chunk(0).get_column(_filter_column_id))) {
      auto val_table = ref_col->get_referenced_table();
      std::vector<std::vector<T>> values = {};
      for (size_t chunk = 0; chunk < val_table->chunk_count(); chunk++) {
        if (auto val_col =
                std::dynamic_pointer_cast<ValueColumn<T>>(val_table->get_chunk(chunk).get_column(_filter_column_id))) {
          values.emplace_back(val_col->get_values());
        } else {
          throw std::logic_error("Referenced table must only contain value columns");
        }
      }
      auto in_pos_list = ref_col->get_pos_list();

      for (size_t pos = 0; pos < in_pos_list->size(); pos++) {
        auto chunk_id = get_chunk_id_from_row_id((*in_pos_list)[pos]);
        auto chunk_offset = get_chunk_offset_from_row_id((*in_pos_list)[pos]);
        if (comp(values[chunk_id][chunk_offset], _filter_value)) {
          _pos_list->emplace_back(get_row_id_from_chunk_id_and_chunk_offset(chunk_id, chunk_offset));
        }
      }
    } else {
      for (ChunkID chunk_id = 0; chunk_id < _table->chunk_count(); ++chunk_id) {
        auto &chunk = _table->get_chunk(chunk_id);
        auto base_column = chunk.get_column(_filter_column_id);
        auto &values = std::dynamic_pointer_cast<ValueColumn<T>>(base_column)->get_values();
        for (ChunkOffset chunk_offset = 0; chunk_offset < chunk.size(); ++chunk_offset) {
          if (comp(values[chunk_offset], _filter_value)) {
            _pos_list->emplace_back(get_row_id_from_chunk_id_and_chunk_offset(chunk_id, chunk_offset));
          }
        }
      }
    }
  }

  void execute() {
    // Definining all possible operators here might appear odd. Chances are, however, that we will not
    // have a similar comparison anywhere else. Index scans, for example, would not use an adaptable binary
    // predicate, but will have to use different methods (lower_range, upper_range, ...) based on the
    // chosen operator. For now, we can save us some dark template magic by using the switch below.
    // DO NOT copy this code, however, without discussing if there is a better way to avoid code duplication.

{'fraction_non_alphanumeric': 0.11358495058935587, 'fraction_numerical': 0.002381235861412073, 'mean_word_length': 3.9557522123893807, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5374129', 'n_tokens_mistral': 2618, 'n_tokens_neox': 2578, 'n_words': 593}