/**
 *    Copyright (C) 2022-present MongoDB, Inc.
 *
 *    This program is free software: you can redistribute it and/or modify
 *    it under the terms of the Server Side Public License, version 1,
 *    as published by MongoDB, Inc.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    Server Side Public License for more details.
 *
 *    You should have received a copy of the Server Side Public License
 *    along with this program. If not, see
 *    <http://www.mongodb.com/licensing/server-side-public-license>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the Server Side Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#include "mongo/db/query/optimizer/utils/abt_hash.h"

#include <cstdint>
#include <functional>
#include <set>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

#include <absl/container/node_hash_map.h>
#include <absl/hash/hash.h>
#include <boost/optional/optional.hpp>

#include "mongo/db/exec/sbe/makeobj_spec.h"
#include "mongo/db/exec/sbe/values/value.h"
#include "mongo/db/query/optimizer/algebra/operator.h"
#include "mongo/db/query/optimizer/algebra/polyvalue.h"
#include "mongo/db/query/optimizer/comparison_op.h"
#include "mongo/db/query/optimizer/defs.h"
#include "mongo/db/query/optimizer/node.h"  // IWYU pragma: keep
#include "mongo/db/query/optimizer/syntax/expr.h"
#include "mongo/db/query/optimizer/syntax/path.h"
#include "mongo/db/query/optimizer/utils/strong_alias.h"
#include "mongo/db/query/optimizer/utils/utils.h"
#include "mongo/util/assert_util.h"

namespace mongo::optimizer {

static size_t computeCollationHash(const ProjectionCollationSpec& spec) {
    size_t collationHash = 17;
    for (const auto& entry : spec) {
        updateHash(collationHash, ProjectionName::Hasher()(entry.first));
        updateHash(collationHash, std::hash<CollationOp>()(entry.second));
    }
    return collationHash;
}

static size_t computePropertyProjectionsHash(const ProjectionNameVector& projections) {
    size_t resultHash = 17;
    for (const ProjectionName& projection : projections) {
        updateHashUnordered(resultHash, ProjectionName::Hasher()(projection));
    }
    return resultHash;
}

static size_t computeProjectionRequirementHash(
    const ProjectionNameOrderPreservingSet& projections) {
    return computePropertyProjectionsHash(projections.getVector());
}

static size_t computeDistributionHash(const DistributionRequirement& prop) {
    size_t resultHash = 17;
    const auto& distribAndProjections = prop.getDistributionAndProjections();
    updateHash(resultHash, std::hash<DistributionType>()(distribAndProjections._type));
    updateHash(resultHash, computePropertyProjectionsHash(distribAndProjections._projectionNames));
    return resultHash;
}

/**
 * Hasher for ABT nodes. Used in conjunction with memo.
 */
class ABTHashTransporter {
public:
    /**
     * Nodes
     */
    template <typename T, typename... Ts>
    size_t transport(const T& /*node*/, Ts&&...) {
        // Physical nodes do not currently need to implement hash.
        static_assert(!std::is_base_of_v<PathSyntaxSort, T> &&
                          !std::is_base_of_v<ExpressionSyntaxSort, T> && !canBeLogicalNode<T>(),
                      "Logical nodes, paths, and expressions must implement their hash.");
        uasserted(6624142, "must implement custom hash");
    }

    size_t transport(const References& references, std::vector<size_t> inResults) {
        return computeHashSeq<1>(computeVectorHash(inResults));
    }

    size_t transport(const ExpressionBinder& binders, std::vector<size_t> inResults) {
        return computeHashSeq<2>(
            computeVectorHash<ProjectionName, ProjectionName::Hasher>(binders.names()),
            computeVectorHash(inResults));
    }

    size_t transport(const ScanNode& node, size_t bindResult) {
        return computeHashSeq<3>(std::hash<std::string>()(node.getScanDefName()), bindResult);
    }

    size_t transport(const ValueScanNode& node, size_t bindResult) {
        // Specifically not hashing props here. Those are compared in the equality operator.
        return computeHashSeq<46>(std::hash<bool>()(node.getHasRID()),
                                  std::hash<size_t>()(node.getArraySize()),
                                  ABTHashGenerator::generate(node.getValueArray()),
                                  bindResult);
    }

    size_t transport(const FilterNode& node, size_t childResult, size_t filterResult) {
        return computeHashSeq<5>(filterResult, childResult);
    }

    size_t transport(const EvaluationNode& node, size_t childResult, size_t projectionResult) {
        return computeHashSeq<6>(projectionResult, childResult);
    }

    size_t transport(const RIDIntersectNode& node,
                     size_t leftChildResult,
                     size_t rightChildResult) {
        // Specifically always including children.
        return computeHashSeq<45>(ProjectionName::Hasher()(node.getScanProjectionName()),
                                  leftChildResult,
                                  rightChildResult);
    }

    size_t transport(const RIDUnionNode& node,
                     size_t leftChildResult,
                     size_t rightChildResult,
                     size_t bindResult,
                     size_t refsResult) {
        return computeHashSeq<47>(ProjectionName::Hasher()(node.getScanProjectionName()),
                                  leftChildResult,
                                  rightChildResult,
                                  bindResult,
                                  refsResult);
    }

    size_t transport(const BinaryJoinNode& node,
                     size_t leftChildResult,
                     size_t rightChildResult,
                     size_t filterResult) {
        // Specifically always including children.
        return computeHashSeq<7>(filterResult, leftChildResult, rightChildResult);
    }

    size_t transport(const UnionNode& node,
                     std::vector<size_t> childResults,
                     size_t bindResult,
                     size_t refsResult) {
        // Specifically always including children.
        return computeHashSeq<9>(bindResult, refsResult, computeVectorHash(childResults));
    }

    size_t transport(const GroupByNode& node,
                     size_t childResult,
                     size_t bindAggResult,
                     size_t refsAggResult,
                     size_t bindGbResult,
                     size_t refsGbResult) {
        return computeHashSeq<10>(bindAggResult,
                                  refsAggResult,
                                  bindGbResult,
                                  refsGbResult,
                                  std::hash<GroupNodeType>()(node.getType()),
                                  childResult);
    }

    size_t transport(const UnwindNode& node,
                     size_t childResult,
                     size_t bindResult,
                     size_t refsResult) {
        return computeHashSeq<11>(
            std::hash<bool>()(node.getRetainNonArrays()), bindResult, refsResult, childResult);
    }

    size_t transport(const CollationNode& node, size_t childResult, size_t /*refsResult*/) {
        return computeHashSeq<13>(computeCollationHash(node.getCollationSpec()), childResult);
    }

    size_t transport(const LimitSkipNode& node, size_t childResult) {
        size_t limitSkipHash = 17;
        updateHash(limitSkipHash, std::hash<int64_t>()(node.getLimit()));
        updateHash(limitSkipHash, std::hash<int64_t>()(node.getSkip()));
        return computeHashSeq<14>(limitSkipHash, childResult);
    }

    size_t transport(const ExchangeNode& node, size_t childResult, size_t /*refsResult*/) {
        return computeHashSeq<43>(computeDistributionHash(node.getProperty()), childResult);
    }

    size_t transport(const RootNode& node, size_t childResult, size_t /*refsResult*/) {
        return computeHashSeq<15>(computeProjectionRequirementHash(node.getProjections()),
                                  childResult);
    }

    /**
     * Expressions
     */
    size_t transport(const Blackhole& expr) {
        return computeHashSeq<16>();
    }

    size_t transport(const Constant& expr) {
        auto [tag, val] = expr.get();
        // Extended types cannot be used directly with hashValue().
        if (tag != sbe::value::TypeTags::Nothing && sbe::value::tagToType(tag) == BSONType::EOO) {
            if (tag == sbe::value::TypeTags::makeObjSpec) {
                return computeHashSeq<17>(
                    absl::Hash<sbe::MakeObjSpec>{}(*sbe::value::getMakeObjSpecView(val)));
            }

            // Extended types must implement their own hash functions.
            MONGO_UNREACHABLE_TASSERT(7936706);
        }
        return computeHashSeq<17>(sbe::value::hashValue(tag, val));
    }

    size_t transport(const Variable& expr) {
        return computeHashSeq<18>(ProjectionName::Hasher()(expr.name()));
    }

    size_t transport(const UnaryOp& expr, size_t inResult) {
        return computeHashSeq<19>(std::hash<Operations>()(expr.op()), inResult);
    }

    size_t transport(const BinaryOp& expr, size_t leftResult, size_t rightResult) {
        return computeHashSeq<20>(std::hash<Operations>()(expr.op()), leftResult, rightResult);
    }

    size_t transport(const If& expr, size_t condResult, size_t thenResult, size_t elseResult) {
        return computeHashSeq<21>(condResult, thenResult, elseResult);
    }

    size_t transport(const Let& expr, size_t bindResult, size_t exprResult) {
        return computeHashSeq<22>(ProjectionName::Hasher()(expr.varName()), bindResult, exprResult);
    }

    size_t transport(const LambdaAbstraction& expr, size_t inResult) {
        return computeHashSeq<23>(ProjectionName::Hasher()(expr.varName()), inResult);
    }

    size_t transport(const LambdaApplication& expr, size_t lambdaResult, size_t argumentResult) {
        return computeHashSeq<24>(lambdaResult, argumentResult);
    }

    size_t transport(const FunctionCall& expr, std::vector<size_t> argResults) {
        return computeHashSeq<25>(std::hash<std::string>()(expr.name()),
                                  computeVectorHash(argResults));
    }

    size_t transport(const EvalPath& expr, size_t pathResult, size_t inputResult) {
        return computeHashSeq<26>(pathResult, inputResult);
    }

    size_t transport(const EvalFilter& expr, size_t pathResult, size_t inputResult) {
        return computeHashSeq<27>(pathResult, inputResult);
    }

    size_t transport(const Source& expr) {
        return computeHashSeq<28>();
    }

    /**
     * Paths
     */
    size_t transport(const PathConstant& path, size_t inResult) {
        return computeHashSeq<29>(inResult);
    }

    size_t transport(const PathLambda& path, size_t inResult) {
        return computeHashSeq<30>(inResult);
    }

    size_t transport(const PathIdentity& path) {
        return computeHashSeq<31>();
    }

    size_t transport(const PathDefault& path, size_t inResult) {
        return computeHashSeq<32>(inResult);
    }

    size_t transport(const PathCompare& path, size_t valueResult) {
        return computeHashSeq<33>(std::hash<Operations>()(path.op()), valueResult);
    }

    size_t transport(const PathDrop& path) {
        size_t namesHash = 17;
        for (const FieldNameType& name : path.getNames()) {
            updateHash(namesHash, FieldNameType::Hasher()(name));
        }
        return computeHashSeq<34>(namesHash);
    }

    size_t transport(const PathKeep& path) {
        size_t namesHash = 17;
        for (const FieldNameType& name : path.getNames()) {
            updateHash(namesHash, FieldNameType::Hasher()(name));
        }
        return computeHashSeq<35>(namesHash);
    }

    size_t transport(const PathObj& path) {
        return computeHashSeq<36>();
    }

    size_t transport(const PathArr& path) {
        return computeHashSeq<37>();
    }

    size_t transport(const PathTraverse& path, size_t inResult) {
        return computeHashSeq<38>(inResult, std::hash<size_t>()(path.getMaxDepth()));
    }

    size_t transport(const PathField& path, size_t inResult) {
        return computeHashSeq<39>(FieldNameType::Hasher()(path.name()), inResult);
    }

    size_t transport(const PathGet& path, size_t inResult) {
        return computeHashSeq<40>(FieldNameType::Hasher()(path.name()), inResult);
    }

    size_t transport(const PathComposeM& path, size_t leftResult, size_t rightResult) {
        return computeHashSeq<41>(leftResult, rightResult);
    }

    size_t transport(const PathComposeA& path, size_t leftResult, size_t rightResult) {
        return computeHashSeq<42>(leftResult, rightResult);
    }

    size_t generate(const ABT& node) {
        return algebra::transport<false>(node, *this);
    }

    size_t generate(const ABT::reference_type& nodeRef) {
        return algebra::transport<false>(nodeRef, *this);
    }
};

size_t ABTHashGenerator::generate(const ABT& node) {
    ABTHashTransporter gen;
    return gen.generate(node);
}

size_t ABTHashGenerator::generate(const ABT::reference_type& nodeRef) {
    ABTHashTransporter gen;
    return gen.generate(nodeRef);
}

}  // namespace mongo::optimizer
