// Copyright 2019, University of Freiburg,
// Chair of Algorithms and Data Structures.
// Author: Florian Kramer (florian.kramer@neptun.uni-freiburg.de)
//         Johannes Herrmann (johannes.r.herrmann(at)gmail.com)

#ifndef QLEVER_REDUCED_FEATURE_SET_FOR_CPP17

#ifndef QLEVER_SRC_ENGINE_TRANSITIVEPATHBASE_H
#define QLEVER_SRC_ENGINE_TRANSITIVEPATHBASE_H

#include <absl/hash/hash.h>

#include <memory>

#include "backports/functional.h"
#include "engine/Operation.h"
#include "engine/QueryExecutionTree.h"

using TreeAndCol = std::pair<std::shared_ptr<QueryExecutionTree>, size_t>;
struct TransitivePathSide {
  // treeAndCol contains the QueryExecutionTree of this side and the column
  // where the Ids of this side are located. This member only has a value if
  // this side was bound.
  std::optional<TreeAndCol> treeAndCol_;
  // Column of the sub table where the Ids of this side are located
  size_t subCol_;
  TripleComponent value_;
  // The column in the output table where this side Ids are written to.
  // This member is set by the TransitivePath class
  size_t outputCol_ = 0;

  bool isVariable() const { return value_.isVariable(); }

  bool isBoundVariable() const { return treeAndCol_.has_value(); }

  bool isUnboundVariable() const { return isVariable() && !isBoundVariable(); }

  std::string getCacheKey() const {
    std::ostringstream os;
    if (!isVariable()) {
      os << "Value " << value_;
    }

    os << ", subColumn: " << subCol_ << "to " << outputCol_;

    if (treeAndCol_.has_value()) {
      const auto& [tree, col] = treeAndCol_.value();
      os << ", Subtree:\n";
      os << tree->getCacheKey() << "with join column " << col << "\n";
    }
    return std::move(os).str();
  }

  bool isSortedOnInputCol() const {
    if (!treeAndCol_.has_value()) {
      return false;
    }

    auto [tree, col] = treeAndCol_.value();
    const std::vector<ColumnIndex>& sortedOn =
        tree->getRootOperation()->getResultSortedOn();

    // If the column contains undef values, we can't guarantee the output order.
    return !sortedOn.empty() && sortedOn[0] == col &&
           tree->getVariableAndInfoByColumnIndex(col)
                   .second.mightContainUndef_ ==
               ColumnIndexAndTypeInfo::UndefStatus::AlwaysDefined;
  }

  TransitivePathSide clone() const {
    TransitivePathSide copy = *this;
    if (copy.treeAndCol_.has_value()) {
      copy.treeAndCol_.value().first = copy.treeAndCol_.value().first->clone();
    }
    return copy;
  }
};

// We deliberately use the `std::` variants of a hash set because `absl`s types
// are not exception safe.
using Set = std::unordered_set<Id, absl::Hash<Id>, std::equal_to<Id>,
                               ad_utility::AllocatorWithLimit<Id>>;
using SetWithGraph =
    std::unordered_set<std::pair<Id, Id>, absl::Hash<std::pair<Id, Id>>,
                       std::equal_to<std::pair<Id, Id>>,
                       ad_utility::AllocatorWithLimit<std::pair<Id, Id>>>;

// Alias for common type
using PayloadTable = std::optional<IdTableView<0>>;

// Helper struct, that allows a generator to yield a a node and all its
// connected nodes (the `targets`), along with a local vocabulary and the row
// index of the node in the input table. The `IdTable` pointer might be null if
// the `Id` is not associated with a table. In this case the `row` value does
// not represent anything meaningful and should not be used.
struct NodeWithTargets {
  Id node_;
  // Graph id, undefined if no graph is set.
  Id graph_;
  // Set of target ids.
  Set targets_;
  LocalVocab localVocab_;
  PayloadTable idTable_;
  // Corresponding row in `idTable_`.
  size_t row_;

  // Explicit to prevent issues with co_yield and lifetime.
  // See https://gcc.gnu.org/bugzilla/show_bug.cgi?id=103909 for more info.
  NodeWithTargets(Id node, Id graph, Set targets, LocalVocab localVocab,
                  PayloadTable idTable, size_t row)
      : node_{node},
        graph_{graph},
        targets_{std::move(targets)},
        localVocab_{std::move(localVocab)},
        idTable_{std::move(idTable)},
        row_{row} {}
};

using NodeGenerator = cppcoro::generator<NodeWithTargets>;

/**
 * @class TransitivePathBase
 * @brief A common base class for different implementations of the Transitive
 * Path operation. To create an actual object use the static factory function
 * `makeStaticPath`.
 *
 */
class TransitivePathBase : public Operation {
 protected:
  using Graphs = ScanSpecificationAsTripleComponent::GraphFilter;

  std::shared_ptr<QueryExecutionTree> subtree_;
  TransitivePathSide lhs_;
  TransitivePathSide rhs_;
  size_t resultWidth_ = 2;
  size_t minDist_;
  size_t maxDist_;
  VariableToColumnMap variableColumns_;
  // Indicate that the variable is only bound because the path is empty, not
  // because `bindLeftOrRightSide` was called. This means that it is bound to a
  // full scan of all subjects and objects in the knowledge graph, but can be
  // re-bound to something cheaper later if the query permits it.
  bool boundVariableIsForEmptyPath_ = false;

  // Store the active graphs for the transitive path operation. This is used to
  // correctly match against the proper graph when the minimum distance is 0.
  Graphs activeGraphs_;
  std::optional<Variable> graphVariable_;

  // Helper variable to allow joining a graph column without name clashes in
  // case the graph variable is the same as the join variable when the graph is
  // bound in `minDist_ == 0` scenarios. This is necessary for queries of the
  // form `?x wdt:P31 ?g . GRAPH ?g { ?g wdt:P279* ?y }`.
  static const Variable internalGraphHelper_;

 public:
  TransitivePathBase(QueryExecutionContext* qec,
                     std::shared_ptr<QueryExecutionTree> child,
                     TransitivePathSide leftSide, TransitivePathSide rightSide,
                     size_t minDist, size_t maxDist, Graphs activeGraphs,
                     const std::optional<Variable>& graphVariable);

  ~TransitivePathBase() override = 0;

  /**
   * Returns a new TransitivePath operation that uses the fact that leftop
   * generates all possible values for the left side of the paths. If the
   * results of leftop is smaller than all possible values this will result in a
   * faster transitive path operation (as the transitive paths has to be
   * computed for fewer elements).
   */
  std::shared_ptr<TransitivePathBase> bindLeftSide(
      std::shared_ptr<QueryExecutionTree> leftop, size_t inputCol) const;

  /**
   * Returns a new TransitivePath operation that uses the fact that rightop
   * generates all possible values for the right side of the paths. If the
   * results of rightop is smaller than all possible values this will result in
   * a faster transitive path operation (as the transitive paths has to be
   * computed for fewer elements).
   */
  std::shared_ptr<TransitivePathBase> bindRightSide(
      std::shared_ptr<QueryExecutionTree> rightop, size_t inputCol) const;

  bool isBoundOrId() const;

  /**
   * Getters, mainly necessary for testing
   */
  size_t getMinDist() const { return minDist_; }
  size_t getMaxDist() const { return maxDist_; }
  const TransitivePathSide& getLeft() const { return lhs_; }
  const TransitivePathSide& getRight() const { return rhs_; }

 protected:
  std::string getCacheKeyImpl() const override;

  /**
   * @brief Decide on which transitive path side the hull computation should
   * start and where it should end. The start and target side are chosen by
   * the following criteria:
   *
   * 1. If a side is bound, then this side will be the start side.
   * 2. If a side is an id, then this side will be the start side.
   * 3. If both sides are variables, the left side is chosen as start
   * (arbitrarily).
   *
   * @return std::pair<TransitivePathSide&, TransitivePathSide&> The first entry
   * of the pair is the start side, the second entry is the target side.
   */
  std::pair<TransitivePathSide&, TransitivePathSide&> decideDirection();

  /**
   * @brief Fill the given table with the transitive hull and use the
   * startSideTable to fill in the rest of the columns.
   *
   * @param hull The transitive hull, represented by a generator that yields
   * sets of connected nodes with some metadata.
   * @param startSideCol The column of the result table for the startSide of the
   * hull
   * @param targetSideCol The column of the result table for the targetSide of
   * the hull
   * @param yieldOnce If true, the generator will yield only a single time.
   * @param inputWidth The width of the input table that is referenced by the
   * elements of `hull`.
   */
  Result::Generator fillTableWithHull(NodeGenerator hull, size_t startSideCol,
                                      size_t targetSideCol, bool yieldOnce,
                                      size_t inputWidth) const;

  // Copy the columns from the input table to the output table
  template <size_t INPUT_WIDTH, size_t OUTPUT_WIDTH>
  void copyColumns(const IdTableView<INPUT_WIDTH>& inputTable,
                   IdTableStatic<OUTPUT_WIDTH>& outputTable, size_t inputRow,
                   size_t outputRow) const;

  // Return the actual index of the graph column in `tree`. If
  // `internalGraphHelper_` is present it takes precedence over
  // `graphVariable_`. If this instance does not use graph variables, or none of
  // the variables is present in the `tree`, return `std::nullopt`;
  std::optional<ColumnIndex> getActualGraphColumnIndex(
      const std::shared_ptr<QueryExecutionTree>& tree) const;

  // Return how many columns would be joined given the passed `tree`. Return 1
  // if `getActualGraphColumnIndex(tree)` is `std::nullopt` or the returned
  // index is equal to `joinColumn`. Return 2 otherwise.
  size_t numJoinColumnsWith(const std::shared_ptr<QueryExecutionTree>& tree,
                            ColumnIndex joinColumn) const;

 public:
  std::string getDescriptor() const override;

  size_t getResultWidth() const override;

  std::vector<ColumnIndex> resultSortedOn() const override;

  bool knownEmptyResult() override;

  float getMultiplicity(size_t col) override;

 private:
  uint64_t getSizeEstimateBeforeLimit() override;

  template <size_t INPUT_WIDTH, size_t OUTPUT_WIDTH>
  Result::Generator fillTableWithHullImpl(NodeGenerator hull,
                                          size_t startSideCol,
                                          size_t targetSideCol,
                                          bool yieldOnce) const;

  // Create two instances of `IndexScan` that perform full scans using the
  // pattern `?variable ?a ?b` and `?c ?d ?variable`, respectively and properly
  // use `activeGraphs` and the optional `graphVariable`.
  static std::array<std::shared_ptr<QueryExecutionTree>, 2> makeIndexScanPair(
      QueryExecutionContext* qec, Graphs activeGraphs, const Variable& variable,
      const std::optional<Variable>& graphVariable);

  // Return an execution tree, that "joins" the given `tripleComponent` with all
  // the subjects or objects in the knowledge graph, so if the graph does not
  // contain this value it is filtered out.
  static std::shared_ptr<QueryExecutionTree> joinWithIndexScan(
      QueryExecutionContext* qec, Graphs activeGraphs,
      const std::optional<Variable>& graphVariable,
      const TripleComponent& tripleComponent);

  // Return an execution tree that represents one side of an empty path. This is
  // used as a starting point for evaluating the empty path and returns a single
  // column containung all distinct entities the appear either as a subject or
  // object in the knowledge graph. The optional parameter `variable` can be set
  // to explicitly define the name of the column this produces (useful for
  // subsequent joins), by default it is `?internal_property_path_variable_x`.
  static std::shared_ptr<QueryExecutionTree> makeEmptyPathSide(
      QueryExecutionContext* qec, Graphs activeGraphs,
      const std::optional<Variable>& graphVariable,
      std::optional<Variable> variable = std::nullopt);

  // Make sure that all values in `inputCol` returned by `leftOrRightOp` can be
  // found in the knowledge graph. In many cases we can statically guarantee
  // this and just return the `leftOrRightOp` unchanged, in all other cases the
  // result will be a join with the result of `makeEmptyPathSide` above.
  std::shared_ptr<QueryExecutionTree> matchWithKnowledgeGraph(
      size_t& inputCol,
      std::shared_ptr<QueryExecutionTree> leftOrRightOp) const;

 public:
  size_t getCostEstimate() override;

  /**
   * @brief Make a concrete TransitivePath object using the given parameters.
   * The concrete object will either be TransitivePathFallback or
   * TransitivePathBinSearch, depending on the useBinSearch flag.
   *
   * @param qec QueryExecutionContext for the TransitivePath Operation
   * @param child QueryExecutionTree for the subquery of the TransitivePath
   * @param leftSide Settings for the left side of the TransitivePath
   * @param rightSide Settings for the right side of the TransitivePath
   * @param minDist Minimum distance a resulting path may have (distance =
   * number of nodes)
   * @param maxDist Maximum distance a resulting path may have (distance =
   * number of nodes)
   * @param useBinSearch If true, the returned object will be a
   * TransitivePathBinSearch. Else it will be a TransitivePathFallback
   * @param activeGraphs Contains the graphs that are active in the current
   * context.
   * @param graphVariable Set a graph variable when inside a `GRAPH ?g { ... }`.
   */
  static std::shared_ptr<TransitivePathBase> makeTransitivePath(
      QueryExecutionContext* qec, std::shared_ptr<QueryExecutionTree> child,
      TransitivePathSide leftSide, TransitivePathSide rightSide, size_t minDist,
      size_t maxDist, bool useBinSearch, Graphs activeGraphs = Graphs::All(),
      const std::optional<Variable>& graphVariable = std::nullopt);

  /**
   * @brief Make a concrete TransitivePath object using the given parameters.
   * The concrete object will either be TransitivePathFallback or
   * TransitivePathBinSearch, depending on the runtime constant "use-binsearch".
   *
   * @param qec QueryExecutionContext for the TransitivePath Operation
   * @param child QueryExecutionTree for the subquery of the TransitivePath
   * @param leftSide Settings for the left side of the TransitivePath
   * @param rightSide Settings for the right side of the TransitivePath
   * @param minDist Minimum distance a resulting path may have (distance =
   * number of nodes)
   * @param maxDist Maximum distance a resulting path may have (distance =
   * number of nodes)
   * @param activeGraphs Contains the graphs that are active in the current
   * context.
   * @param graphVariable Set a graph variable when inside a `GRAPH ?g { ... }`.
   */
  static std::shared_ptr<TransitivePathBase> makeTransitivePath(
      QueryExecutionContext* qec, std::shared_ptr<QueryExecutionTree> child,
      TransitivePathSide leftSide, TransitivePathSide rightSide, size_t minDist,
      size_t maxDist, Graphs activeGraphs = Graphs::All(),
      const std::optional<Variable>& graphVariable = std::nullopt);

  std::vector<QueryExecutionTree*> getChildren() override;

  VariableToColumnMap computeVariableToColumnMap() const override;

  bool columnOriginatesFromGraphOrUndef(
      const Variable& variable) const override;

  // The internal implementation of `bindLeftSide` and `bindRightSide` which
  // share a lot of code.
  std::shared_ptr<TransitivePathBase> bindLeftOrRightSide(
      std::shared_ptr<QueryExecutionTree> leftOrRightOp, size_t inputCol,
      bool isLeft) const;

  // Return a set of subtrees that can be used alternatively when the left or
  // right side is bound. This is used by the `TransitivePathBinSearch` class,
  // which has to store both ways to sort the subtree until it knows which side
  // becomes bound.
  virtual ql::span<const std::shared_ptr<QueryExecutionTree>>
  alternativeSubtrees() const {
    return {};
  }

  // For an unbound transitive path with a graph variable, return the column
  // index of the graph variable. For a bound transitive path the return value
  // represents the first column of the payload. The graph follows after that
  // (if present).
  static size_t firstGraphOrPayloadColumnIndex() { return 2; }
};

#endif  // QLEVER_SRC_ENGINE_TRANSITIVEPATHBASE_H

#endif
