/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 **/

#ifndef QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_
#define QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_

#include <map>
#include <unordered_map>
#include <utility>
#include <vector>

#include "query_execution/QueryContext.hpp"
#include "query_optimizer/QueryPlan.hpp"
#include "query_optimizer/expressions/ExprId.hpp"
#include "query_optimizer/physical/LIPFilterConfiguration.hpp"
#include "query_optimizer/physical/Aggregate.hpp"
#include "query_optimizer/physical/FilterJoin.hpp"
#include "query_optimizer/physical/HashJoin.hpp"
#include "query_optimizer/physical/Physical.hpp"
#include "query_optimizer/physical/Selection.hpp"
#include "utility/Macros.hpp"

#include "glog/logging.h"

namespace quickstep {

namespace serialization {

class QueryContext;
class LIPFilterDeployment;

}

class CatalogAttribute;

namespace optimizer {

/** \addtogroup QueryOptimizer
 *  @{
 */

/**
 * @brief Generates backend LIPFilter deployments from physical plan's information.
 */
class LIPFilterGenerator {
 public:
  /**
   * @brief Constructor.
   *
   * @param lip_filter_configuration The LIPFilter configuration information
   *        generated by physical optimizer.
   */
  explicit LIPFilterGenerator(
      const physical::LIPFilterConfigurationPtr &lip_filter_configuration)
      : lip_filter_configuration_(lip_filter_configuration) {
    DCHECK(lip_filter_configuration_ != nullptr);
  }

  /**
   * @brief Collect the ExprId to CatalogAttribute mapping information for the
   *        given physical node.
   *
   * @param node A physical plan node.
   * @param attribute_substitution_map A map that maps each ExprId to the
   *        backend relation's CatalogAttribute's.
   */
  void registerAttributeMap(
      const physical::PhysicalPtr &node,
      const std::unordered_map<expressions::ExprId, const CatalogAttribute *> &attribute_substitution_map);

  /**
   * @brief Add physical-to-execution mapping information for deploying LIPFilters
   *        to an aggregation.
   *
   * @param aggregate A physical Aggregate node.
   * @param aggregate_operator_index The index of the AggregationOperator that
   *        corresponds to \p aggregate in the execution plan.
   */
  void addAggregateInfo(const physical::AggregatePtr &aggregate,
                        const QueryPlan::DAGNodeIndex aggregate_operator_index) {
    prober_infos_.emplace_back(aggregate, aggregate_operator_index);
  }

  /**
   * @brief Add physical-to-execution mapping information for deploying LIPFilters
   *        to a FilterJoin node.
   *
   * @param filter_join A physical FilterJoin node.
   * @param build_filter_operator_index The index of the BuildLIPFilterOperator
   *        that corresponds to \p filter_join in the execution plan.
   */
  void addFilterJoinInfo(const physical::FilterJoinPtr &filter_join,
                         const QueryPlan::DAGNodeIndex build_filter_operator_index) {
    builder_infos_.emplace_back(filter_join, build_filter_operator_index);
    prober_infos_.emplace_back(filter_join, build_filter_operator_index);
  }

  /**
   * @brief Add physical-to-execution mapping information for deploying LIPFilters
   *        to a hash-join.
   *
   * @param hash_join A physical HashJoin node.
   * @param build_operator_index The index of the BuildHashOperator that corresponds
   *        to \p hash_join in the execution plan.
   * @param join_operator_index The index of the HashJoinOperator that corresponds
   *        to \p hash_join in the execution plan.
   */
  void addHashJoinInfo(const physical::HashJoinPtr &hash_join,
                       const QueryPlan::DAGNodeIndex build_operator_index,
                       const QueryPlan::DAGNodeIndex join_operator_index) {
    builder_infos_.emplace_back(hash_join, build_operator_index);
    prober_infos_.emplace_back(hash_join, join_operator_index);
  }

  /**
   * @brief Add physical-to-execution mapping information for deploying LIPFilters
   *        to a selection.
   *
   * @param selection A physical Selection node.
   * @param select_operator_index The index of the SelectOperator that corresponds
   *        to \p selection in the execution plan.
   */
  void addSelectionInfo(const physical::SelectionPtr &selection,
                        const QueryPlan::DAGNodeIndex select_operator_index) {
    prober_infos_.emplace_back(selection, select_operator_index);
  }

  /**
   * @brief Deploy the LIPFilters to the execution plan.
   *
   * @param execution_plan The execution plan.
   * @param query_context_proto QueryContext protobuf for the execution plan.
   */
  void deployLIPFilters(QueryPlan *execution_plan,
                        serialization::QueryContext *query_context_proto);

 private:
  /**
   * @brief Internal data structure for representing a LIPFilter builder.
   */
  struct BuilderInfo {
    BuilderInfo(const physical::PhysicalPtr &builder_node_in,
                const QueryPlan::DAGNodeIndex builder_operator_index_in)
        : builder_node(builder_node_in),
          builder_operator_index(builder_operator_index_in) {
    }
    const physical::PhysicalPtr builder_node;
    const QueryPlan::DAGNodeIndex builder_operator_index;
  };

  /**
   * @brief Internal data structure for representing a LIPFilter prober.
   */
  struct ProberInfo {
    ProberInfo(const physical::PhysicalPtr &prober_node_in,
               const QueryPlan::DAGNodeIndex prober_operator_index_in)
        : prober_node(prober_node_in),
          prober_operator_index(prober_operator_index_in) {
    }
    const physical::PhysicalPtr prober_node;
    const QueryPlan::DAGNodeIndex prober_operator_index;
  };

  void deployBuilderInternal(QueryPlan *execution_plan,
                             serialization::QueryContext *query_context_proto,
                             const physical::PhysicalPtr &builder_node,
                             const QueryPlan::DAGNodeIndex builder_operator_index,
                             const std::vector<physical::LIPFilterBuildInfoPtr> &build_info_vec);

  void deployProberInteral(QueryPlan *execution_plan,
                           serialization::QueryContext *query_context_proto,
                           const physical::PhysicalPtr &prober_node,
                           const QueryPlan::DAGNodeIndex prober_operator_index,
                           const std::vector<physical::LIPFilterProbeInfoPtr> &probe_info_vec);

  serialization::LIPFilterDeployment* getLIPFilterDeploymentProto(
      const QueryPlan::DAGNodeIndex op_index,
      serialization::QueryContext *query_context_proto);

  const physical::LIPFilterConfigurationPtr lip_filter_configuration_;

  std::vector<BuilderInfo> builder_infos_;
  std::vector<ProberInfo> prober_infos_;

  std::map<physical::PhysicalPtr, std::map<expressions::ExprId, const CatalogAttribute *>> attribute_map_;

  // Maps each LIPFilter's building attribute to the LIPFilter's id in QueryContext
  // as well as the LIPFilter's building relational operator's index.
  std::map<std::pair<expressions::ExprId, physical::PhysicalPtr>,
           std::pair<QueryContext::lip_filter_id, QueryPlan::DAGNodeIndex>> lip_filter_builder_map_;

  // Maps each relational operator's index to the attached LIPFilterDeployment's
  // index and proto.
  std::map<QueryPlan::DAGNodeIndex,
           std::pair<int, serialization::LIPFilterDeployment *>> lip_filter_deployment_protos_;

  DISALLOW_COPY_AND_ASSIGN(LIPFilterGenerator);
};

/** @} */

}  // namespace optimizer
}  // namespace quickstep

#endif /* QUICKSTEP_QUERY_OPTIMIZER_LIP_FILTER_GENERATOR_HPP_ */