#pragma once

#include <openssl/md5.h>
#include <openssl/sha.h>
#include <city.h>
#include <farmhash.h>
#include <metrohash.h>

#include <Poco/ByteOrder.h>

#include <Common/SipHash.h>
#include <DataTypes/DataTypesNumber.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/DataTypeDate.h>
#include <DataTypes/DataTypeDateTime.h>
#include <DataTypes/DataTypeArray.h>
#include <DataTypes/DataTypeFixedString.h>
#include <DataTypes/DataTypeEnum.h>
#include <Columns/ColumnsNumber.h>
#include <Columns/ColumnString.h>
#include <Columns/ColumnConst.h>
#include <Columns/ColumnFixedString.h>
#include <Columns/ColumnArray.h>
#include <Columns/ColumnTuple.h>
#include <Common/HashTable/Hash.h>
#include <Functions/IFunction.h>

#include <ext/range.hpp>


namespace DB
{

/** Hashing functions.
  *
  * Half MD5:
  * halfMD5:     String -> UInt64
  *
  * A faster cryptographic hash function:
  * sipHash64:  String -> UInt64
  *
  * Fast non-cryptographic hash function for strings:
  * cityHash64: String -> UInt64
  *
  * A non-cryptographic hash from a tuple of values of any types (uses cityHash64 for strings and intHash64 for numbers):
  * cityHash64:  any* -> UInt64
  *
  * Fast non-cryptographic hash function from any integer:
  * intHash32:    number -> UInt32
  * intHash64:  number -> UInt64
  *
  */

struct HalfMD5Impl
{
    static UInt64 apply(const char * begin, size_t size)
    {
        union
        {
            unsigned char char_data[16];
            Poco::UInt64 uint64_data;
        } buf;

        MD5_CTX ctx;
        MD5_Init(&ctx);
        MD5_Update(&ctx, reinterpret_cast<const unsigned char *>(begin), size);
        MD5_Final(buf.char_data, &ctx);

        return Poco::ByteOrder::flipBytes(buf.uint64_data);        /// Compatibility with existing code.
    }
};

struct MD5Impl
{
    static constexpr auto name = "MD5";
    enum { length = 16 };

    static void apply(const char * begin, const size_t size, unsigned char * out_char_data)
    {
        MD5_CTX ctx;
        MD5_Init(&ctx);
        MD5_Update(&ctx, reinterpret_cast<const unsigned char *>(begin), size);
        MD5_Final(out_char_data, &ctx);
    }
};

struct SHA1Impl
{
    static constexpr auto name = "SHA1";
    enum { length = 20 };

    static void apply(const char * begin, const size_t size, unsigned char * out_char_data)
    {
        SHA_CTX ctx;
        SHA1_Init(&ctx);
        SHA1_Update(&ctx, reinterpret_cast<const unsigned char *>(begin), size);
        SHA1_Final(out_char_data, &ctx);
    }
};

struct SHA224Impl
{
    static constexpr auto name = "SHA224";
    enum { length = 28 };

    static void apply(const char * begin, const size_t size, unsigned char * out_char_data)
    {
        SHA256_CTX ctx;
        SHA224_Init(&ctx);
        SHA224_Update(&ctx, reinterpret_cast<const unsigned char *>(begin), size);
        SHA224_Final(out_char_data, &ctx);
    }
};

struct SHA256Impl
{
    static constexpr auto name = "SHA256";
    enum { length = 32 };

    static void apply(const char * begin, const size_t size, unsigned char * out_char_data)
    {
        SHA256_CTX ctx;
        SHA256_Init(&ctx);
        SHA256_Update(&ctx, reinterpret_cast<const unsigned char *>(begin), size);
        SHA256_Final(out_char_data, &ctx);
    }
};

struct SipHash64Impl
{
    static UInt64 apply(const char * begin, size_t size)
    {
        return sipHash64(begin, size);
    }
};

struct SipHash128Impl
{
    static constexpr auto name = "sipHash128";
    enum { length = 16 };

    static void apply(const char * begin, const size_t size, unsigned char * out_char_data)
    {
        sipHash128(begin, size, reinterpret_cast<char*>(out_char_data));
    }
};

struct IntHash32Impl
{
    using ReturnType = UInt32;

    static UInt32 apply(UInt64 x)
    {
        /// seed is taken from /dev/urandom. It allows you to avoid undesirable dependencies with hashes in different data structures.
        return intHash32<0x75D9543DE018BF45ULL>(x);
    }
};

struct IntHash64Impl
{
    using ReturnType = UInt64;

    static UInt64 apply(UInt64 x)
    {
        return intHash64(x ^ 0x4CF2D2BAAE6DA887ULL);
    }
};


template <typename Impl, typename Name>
class FunctionStringHash64 : public IFunction
{
public:
    static constexpr auto name = Name::name;
    static FunctionPtr create(const Context & context) { return std::make_shared<FunctionStringHash64>(); };

    String getName() const override
    {
        return name;
    }

    size_t getNumberOfArguments() const override { return 1; }

    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
    {
        if (!typeid_cast<const DataTypeString *>(&*arguments[0]))
            throw Exception("Illegal type " + arguments[0]->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);

        return std::make_shared<DataTypeUInt64>();
    }

    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override
    {
        if (const ColumnString * col_from = typeid_cast<const ColumnString *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            auto col_to = std::make_shared<ColumnUInt64>();
            block.safeGetByPosition(result).column = col_to;

            const typename ColumnString::Chars_t & data = col_from->getChars();
            const typename ColumnString::Offsets_t & offsets = col_from->getOffsets();
            typename ColumnUInt64::Container_t & vec_to = col_to->getData();
            size_t size = offsets.size();
            vec_to.resize(size);

            for (size_t i = 0; i < size; ++i)
                vec_to[i] = Impl::apply(
                    reinterpret_cast<const char *>(&data[i == 0 ? 0 : offsets[i - 1]]),
                    i == 0 ? offsets[i] - 1 : (offsets[i] - 1 - offsets[i - 1]));
        }
        else if (const ColumnConstString * col_from = typeid_cast<const ColumnConstString *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            block.safeGetByPosition(result).column = std::make_shared<ColumnConstUInt64>(
                col_from->size(),
                Impl::apply(col_from->getData().data(), col_from->getData().size()));
        }
        else
            throw Exception("Illegal column " + block.safeGetByPosition(arguments[0]).column->getName()
                    + " of first argument of function " + Name::name,
                ErrorCodes::ILLEGAL_COLUMN);
    }
};


template <typename Impl>
class FunctionStringHashFixedString : public IFunction
{
public:
    static constexpr auto name = Impl::name;
    static FunctionPtr create(const Context & context) { return std::make_shared<FunctionStringHashFixedString>(); };

    String getName() const override
    {
        return name;
    }

    size_t getNumberOfArguments() const override { return 1; }

    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
    {
        if (!typeid_cast<const DataTypeString *>(&*arguments[0]))
            throw Exception("Illegal type " + arguments[0]->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);

        return std::make_shared<DataTypeFixedString>(Impl::length);
    }

    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override
    {
        if (const ColumnString * col_from = typeid_cast<const ColumnString *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            auto col_to = std::make_shared<ColumnFixedString>(Impl::length);
            block.safeGetByPosition(result).column = col_to;

            const typename ColumnString::Chars_t & data = col_from->getChars();
            const typename ColumnString::Offsets_t & offsets = col_from->getOffsets();
            auto & chars_to = col_to->getChars();
            const auto size = offsets.size();
            chars_to.resize(size * Impl::length);

            for (size_t i = 0; i < size; ++i)
                Impl::apply(
                    reinterpret_cast<const char *>(&data[i == 0 ? 0 : offsets[i - 1]]),
                    i == 0 ? offsets[i] - 1 : (offsets[i] - 1 - offsets[i - 1]),
                    &chars_to[i * Impl::length]);
        }
        else if (const ColumnConstString * col_from = typeid_cast<const ColumnConstString *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            const auto & data = col_from->getData();

            String hash(Impl::length, 0);
            Impl::apply(data.data(), data.size(), reinterpret_cast<unsigned char *>(&hash[0]));

            block.safeGetByPosition(result).column = std::make_shared<ColumnConstString>(
                col_from->size(),
                hash,
                std::make_shared<DataTypeFixedString>(Impl::length));
        }
        else
            throw Exception("Illegal column " + block.safeGetByPosition(arguments[0]).column->getName()
                    + " of first argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN);
    }
};


template <typename Impl, typename Name>
class FunctionIntHash : public IFunction
{
public:
    static constexpr auto name = Name::name;
    static FunctionPtr create(const Context & context) { return std::make_shared<FunctionIntHash>(); };

private:
    using ToType = typename Impl::ReturnType;

    template <typename FromType>
    void executeType(Block & block, const ColumnNumbers & arguments, size_t result)
    {
        if (ColumnVector<FromType> * col_from = typeid_cast<ColumnVector<FromType> *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            auto col_to = std::make_shared<ColumnVector<ToType>>();
            block.safeGetByPosition(result).column = col_to;

            const typename ColumnVector<FromType>::Container_t & vec_from = col_from->getData();
            typename ColumnVector<ToType>::Container_t & vec_to = col_to->getData();

            size_t size = vec_from.size();
            vec_to.resize(size);
            for (size_t i = 0; i < size; ++i)
                vec_to[i] = Impl::apply(vec_from[i]);
        }
        else if (ColumnConst<FromType> * col_from = typeid_cast<ColumnConst<FromType> *>(block.safeGetByPosition(arguments[0]).column.get()))
        {
            block.safeGetByPosition(result).column = std::make_shared<ColumnConst<ToType>>(col_from->size(), Impl::apply(col_from->getData()));
        }
        else
            throw Exception("Illegal column " + block.safeGetByPosition(arguments[0]).column->getName()
                    + " of first argument of function " + Name::name,
                ErrorCodes::ILLEGAL_COLUMN);
    }

public:
    String getName() const override
    {
        return name;
    }

    size_t getNumberOfArguments() const override { return 1; }

    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
    {
        if (!arguments[0]->isNumeric())
            throw Exception("Illegal type " + arguments[0]->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);

        return std::make_shared<DataTypeNumber<typename Impl::ReturnType>>();
    }

    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override
    {
        IDataType * from_type = block.safeGetByPosition(arguments[0]).type.get();

        if      (typeid_cast<const DataTypeUInt8 *        >(from_type)) executeType<UInt8    >(block, arguments, result);
        else if (typeid_cast<const DataTypeUInt16 *    >(from_type)) executeType<UInt16>(block, arguments, result);
        else if (typeid_cast<const DataTypeUInt32 *    >(from_type)) executeType<UInt32>(block, arguments, result);
        else if (typeid_cast<const DataTypeUInt64 *    >(from_type)) executeType<UInt64>(block, arguments, result);
        else if (typeid_cast<const DataTypeInt8 *        >(from_type)) executeType<Int8    >(block, arguments, result);
        else if (typeid_cast<const DataTypeInt16 *        >(from_type)) executeType<Int16    >(block, arguments, result);
        else if (typeid_cast<const DataTypeInt32 *        >(from_type)) executeType<Int32    >(block, arguments, result);
        else if (typeid_cast<const DataTypeInt64 *        >(from_type)) executeType<Int64    >(block, arguments, result);
        else if (typeid_cast<const DataTypeDate *        >(from_type)) executeType<UInt16>(block, arguments, result);
        else if (typeid_cast<const DataTypeDateTime *    >(from_type)) executeType<UInt32>(block, arguments, result);
        else
            throw Exception("Illegal type " + block.safeGetByPosition(arguments[0]).type->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
    }
};


template <typename T>
static UInt64 toInteger(T x)
{
    return x;
}

template <>
UInt64 toInteger<Float32>(Float32 x);

template <>
UInt64 toInteger<Float64>(Float64 x);


/** We use hash functions called CityHash, FarmHash, MetroHash.
  * In this regard, this template is named with the words `NeighborhoodHash`.
  */
template <typename Impl>
class FunctionNeighbourhoodHash64 : public IFunction
{
public:
    static constexpr auto name = Impl::name;
    static FunctionPtr create(const Context & context) { return std::make_shared<FunctionNeighbourhoodHash64>(); };

private:
    template <typename FromType, bool first>
    void executeIntType(const IColumn * column, ColumnUInt64::Container_t & vec_to)
    {
        if (const ColumnVector<FromType> * col_from = typeid_cast<const ColumnVector<FromType> *>(column))
        {
            const typename ColumnVector<FromType>::Container_t & vec_from = col_from->getData();
            size_t size = vec_from.size();
            for (size_t i = 0; i < size; ++i)
            {
                UInt64 h = IntHash64Impl::apply(toInteger(vec_from[i]));
                if (first)
                    vec_to[i] = h;
                else
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], h));
            }
        }
        else if (const ColumnConst<FromType> * col_from = typeid_cast<const ColumnConst<FromType> *>(column))
        {
            const UInt64 hash = IntHash64Impl::apply(toInteger(col_from->getData()));
            size_t size = vec_to.size();
            if (first)
            {
                vec_to.assign(size, hash);
            }
            else
            {
                for (size_t i = 0; i < size; ++i)
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], hash));
            }
        }
        else
            throw Exception("Illegal column " + column->getName()
                + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN);
    }

    template <bool first>
    void executeString(const IColumn * column, ColumnUInt64::Container_t & vec_to)
    {
        if (const ColumnString * col_from = typeid_cast<const ColumnString *>(column))
        {
            const typename ColumnString::Chars_t & data = col_from->getChars();
            const typename ColumnString::Offsets_t & offsets = col_from->getOffsets();
            size_t size = offsets.size();

            for (size_t i = 0; i < size; ++i)
            {
                const UInt64 h = Impl::Hash64(
                    reinterpret_cast<const char *>(&data[i == 0 ? 0 : offsets[i - 1]]),
                    i == 0 ? offsets[i] - 1 : (offsets[i] - 1 - offsets[i - 1]));
                if (first)
                    vec_to[i] = h;
                else
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], h));
            }
        }
        else if (const ColumnFixedString * col_from = typeid_cast<const ColumnFixedString *>(column))
        {
            const typename ColumnString::Chars_t & data = col_from->getChars();
            size_t n = col_from->getN();
            size_t size = data.size() / n;
            for (size_t i = 0; i < size; ++i)
            {
                const UInt64 h = Impl::Hash64(reinterpret_cast<const char *>(&data[i * n]), n);
                if (first)
                    vec_to[i] = h;
                else
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], h));
            }
        }
        else if (const ColumnConstString * col_from = typeid_cast<const ColumnConstString *>(column))
        {
            const UInt64 hash = Impl::Hash64(col_from->getData().data(), col_from->getData().size());
            const size_t size = vec_to.size();
            if (first)
            {
                vec_to.assign(size, hash);
            }
            else
            {
                for (size_t i = 0; i < size; ++i)
                {
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], hash));
                }
            }
        }
        else
            throw Exception("Illegal column " + column->getName()
                    + " of first argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN);
    }

    template <bool first>
    void executeArray(const IDataType * type, const IColumn * column, ColumnUInt64::Container_t & vec_to)
    {
        const IDataType * nested_type = &*typeid_cast<const DataTypeArray *>(type)->getNestedType();

        if (const ColumnArray * col_from = typeid_cast<const ColumnArray *>(column))
        {
            const IColumn * nested_column = &col_from->getData();
            const ColumnArray::Offsets_t & offsets = col_from->getOffsets();
            const size_t nested_size = nested_column->size();

            ColumnUInt64::Container_t vec_temp(nested_size);
            executeAny<true>(nested_type, nested_column, vec_temp);

            const size_t size = offsets.size();

            for (size_t i = 0; i < size; ++i)
            {
                const size_t begin = i == 0 ? 0 : offsets[i - 1];
                const size_t end = offsets[i];

                UInt64 h = IntHash64Impl::apply(end - begin);
                if (first)
                    vec_to[i] = h;
                else
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], h));

                for (size_t j = begin; j < end; ++j)
                    vec_to[i] = Impl::Hash128to64(typename Impl::uint128_t(vec_to[i], vec_temp[j]));
            }
        }
        else if (const ColumnConstArray * col_from = typeid_cast<const ColumnConstArray *>(column))
        {
            /// NOTE: here, of course, you can do without the materialization of the column.
            ColumnPtr full_column = col_from->convertToFullColumn();
            executeArray<first>(type, &*full_column, vec_to);
        }
        else
            throw Exception("Illegal column " + column->getName()
                    + " of first argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN);
    }

    template <bool first>
    void executeAny(const IDataType * from_type, const IColumn * icolumn, ColumnUInt64::Container_t & vec_to)
    {
        if      (typeid_cast<const DataTypeUInt8 *        >(from_type)) executeIntType<UInt8,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeUInt16 *        >(from_type)) executeIntType<UInt16,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeUInt32 *        >(from_type)) executeIntType<UInt32,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeUInt64 *        >(from_type)) executeIntType<UInt64,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeInt8 *        >(from_type)) executeIntType<Int8,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeInt16 *        >(from_type)) executeIntType<Int16,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeInt32 *        >(from_type)) executeIntType<Int32,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeInt64 *        >(from_type)) executeIntType<Int64,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeEnum8 *        >(from_type)) executeIntType<Int8,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeEnum16 *        >(from_type)) executeIntType<Int16,        first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeDate *        >(from_type)) executeIntType<UInt16,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeDateTime *    >(from_type)) executeIntType<UInt32,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeFloat32 *    >(from_type)) executeIntType<Float32,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeFloat64 *    >(from_type)) executeIntType<Float64,    first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeString *        >(from_type)) executeString    <            first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeFixedString *>(from_type)) executeString    <            first>(icolumn, vec_to);
        else if (typeid_cast<const DataTypeArray *        >(from_type)) executeArray    <            first>(from_type, icolumn, vec_to);
        else
            throw Exception("Unexpected type " + from_type->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
    }

    void executeForArgument(const IDataType * type, const IColumn * column, ColumnUInt64::Container_t & vec_to, bool & is_first)
    {
        /// Flattening of tuples.
        if (const ColumnTuple * tuple = typeid_cast<const ColumnTuple *>(column))
        {
            const Block & tuple_data = tuple->getData();
            for (size_t i = 0, size = tuple_data.columns(); i < size; ++i)
            {
                const ColumnWithTypeAndName & col = tuple_data.getByPosition(i);
                executeForArgument(col.type.get(), col.column.get(), vec_to, is_first);
            }
        }
        else if (const ColumnConstTuple * tuple = typeid_cast<const ColumnConstTuple *>(column))
        {
            ColumnPtr tuple_of_constants = tuple->convertToTupleOfConstants();
            executeForArgument(type, tuple_of_constants.get(), vec_to, is_first);
        }
        else
        {
            if (is_first)
                executeAny<true>(type, column, vec_to);
            else
                executeAny<false>(type, column, vec_to);
        }

        is_first = false;
    }

public:
    String getName() const override
    {
        return name;
    }

    bool isVariadic() const override { return true; }
    size_t getNumberOfArguments() const override { return 0; }

    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
    {
        return std::make_shared<DataTypeUInt64>();
    }

    void executeImpl(Block & block, const ColumnNumbers & arguments, size_t result) override
    {
        size_t rows = block.rows();
        auto col_to = std::make_shared<ColumnUInt64>(rows);
        block.safeGetByPosition(result).column = col_to;

        ColumnUInt64::Container_t & vec_to = col_to->getData();

        if (arguments.empty())
        {
            /// Constant random number from /dev/urandom is used as a hash value of empty list of arguments.
            vec_to.assign(rows, 0xe28dbde7fe22e41c);
        }

        /// The function supports arbitary number of arguments of arbitary types.

        bool is_first_argument = true;
        for (size_t i = 0; i < arguments.size(); ++i)
        {
            const ColumnWithTypeAndName & col = block.safeGetByPosition(arguments[i]);
            executeForArgument(col.type.get(), col.column.get(), vec_to, is_first_argument);
        }

        /// If all arguments are constants, we should return constant result.

        bool all_constants = true;
        for (size_t arg_idx : arguments)
        {
            if (!block.getByPosition(arg_idx).column->isConst())
            {
                all_constants = false;
                break;
            }
        }

        if (all_constants && block.rows() > 0)
            block.getByPosition(result).column = block.getByPosition(result).type->createConstColumn(1, (*block.getByPosition(result).column)[0]);
    }
};


struct URLHashImpl
{
    static UInt64 apply(const char * data, const std::size_t size)
    {
        /// do not take last slash, '?' or '#' character into account
        if (size > 0 && (data[size - 1] == '/' || data[size - 1] == '?' || data[size - 1] == '#'))
            return CityHash64(data, size - 1);

        return CityHash64(data, size);
    }
};


struct URLHierarchyHashImpl
{
    static std::size_t findLevelLength(const UInt64 level, const char * begin, const char * const end)
    {
        auto pos = begin;

        /// Let's parse everything that goes before the path

        /// Suppose that the protocol has already been changed to lowercase.
        while (pos < end && ((*pos > 'a' && *pos < 'z') || (*pos > '0' && *pos < '9')))
            ++pos;

        /** We will calculate the hierarchy only for URLs in which there is a protocol, and after it there are two slashes.
        *    (http, file - fit, mailto, magnet - do not fit), and after two slashes there is still something
        *    For the rest, simply return the full URL as the only element of the hierarchy.
        */
        if (pos == begin || pos == end || !(*pos++ == ':' && pos < end && *pos++ == '/' && pos < end && *pos++ == '/' && pos < end))
        {
            pos = end;
            return 0 == level ? pos - begin : 0;
        }

        /// The domain for simplicity is everything that after the protocol and the two slashes, until the next slash or before `?` or `#`
        while (pos < end && !(*pos == '/' || *pos == '?' || *pos == '#'))
            ++pos;

        if (pos != end)
            ++pos;

        if (0 == level)
            return pos - begin;

        UInt64 current_level = 0;

        while (current_level != level && pos < end)
        {
            /// We go to the next `/` or `?` or `#`, skipping all at the beginning.
            while (pos < end && (*pos == '/' || *pos == '?' || *pos == '#'))
                ++pos;
            if (pos == end)
                break;
            while (pos < end && !(*pos == '/' || *pos == '?' || *pos == '#'))
                ++pos;

            if (pos != end)
                ++pos;

            ++current_level;
        }

        return current_level == level ? pos - begin : 0;
    }

    static UInt64 apply(const UInt64 level, const char * data, const std::size_t size)
    {
        return URLHashImpl::apply(data, findLevelLength(level, data, data + size));
    }
};


class FunctionURLHash : public IFunction
{
public:
    static constexpr auto name = "URLHash";
    static FunctionPtr create(const Context &) { return std::make_shared<FunctionURLHash>(); }

    String getName() const override { return name; }

    bool isVariadic() const override { return true; }
    size_t getNumberOfArguments() const override { return 0; }

    DataTypePtr getReturnTypeImpl(const DataTypes & arguments) const override
    {
        const auto arg_count = arguments.size();
        if (arg_count != 1 && arg_count != 2)
            throw Exception{
                "Number of arguments for function " + getName() + " doesn't match: passed " +
                    toString(arg_count) + ", should be 1 or 2.",
                ErrorCodes::NUMBER_OF_ARGUMENTS_DOESNT_MATCH
            };

        const auto first_arg = arguments.front().get();
        if (!typeid_cast<const DataTypeString *>(first_arg))
            throw Exception{
                "Illegal type " + first_arg->getName() + " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT
            };

        if (arg_count == 2)
        {
            const auto second_arg = arguments.back().get();
            if (!typeid_cast<const DataTypeUInt8 *>(second_arg) &&
                !typeid_cast<const DataTypeUInt16 *>(second_arg) &&
                !typeid_cast<const DataTypeUInt32 *>(second_arg) &&
                !typeid_cast<const DataTypeUInt64 *>(second_arg) &&
                !typeid_cast<const DataTypeInt8 *>(second_arg) &&
                !typeid_cast<const DataTypeInt16 *>(second_arg) &&
                !typeid_cast<const DataTypeInt32 *>(second_arg) &&
                !typeid_cast<const DataTypeInt64 *>(second_arg))
                throw Exception{
                    "Illegal type " + second_arg->getName() + " of argument of function " + getName(),
                    ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT
                };
        }

        return std::make_shared<DataTypeUInt64>();
    }

    void executeImpl(Block & block, const ColumnNumbers & arguments, const size_t result) override
    {
        const auto arg_count = arguments.size();

        if (arg_count == 1)
            executeSingleArg(block, arguments, result);
        else if (arg_count == 2)
            executeTwoArgs(block, arguments, result);
        else
            throw std::logic_error{"got into IFunction::execute with unexpected number of arguments"};
    }

private:
    void executeSingleArg(Block & block, const ColumnNumbers & arguments, const std::size_t result) const
    {
        const auto col_untyped = block.safeGetByPosition(arguments.front()).column.get();

        if (const auto col_from = typeid_cast<const ColumnString *>(col_untyped))
        {
            const auto size = col_from->size();
            const auto col_to = std::make_shared<ColumnUInt64>(size);
            block.safeGetByPosition(result).column = col_to;

            const auto & chars = col_from->getChars();
            const auto & offsets = col_from->getOffsets();
            auto & out = col_to->getData();

            for (const auto i : ext::range(0, size))
                out[i] = URLHashImpl::apply(
                    reinterpret_cast<const char *>(&chars[i == 0 ? 0 : offsets[i - 1]]),
                    i == 0 ? offsets[i] - 1 : (offsets[i] - 1 - offsets[i - 1]));
        }
        else if (const auto col_from = typeid_cast<const ColumnConstString *>(col_untyped))
        {
            block.safeGetByPosition(result).column = std::make_shared<ColumnConstUInt64>(
                col_from->size(),
                URLHashImpl::apply(col_from->getData().data(), col_from->getData().size()));
        }
        else
            throw Exception{
                "Illegal column " + block.safeGetByPosition(arguments[0]).column->getName() +
                " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN};
    }

    void executeTwoArgs(Block & block, const ColumnNumbers & arguments, const std::size_t result) const
    {
        const auto level_col = block.safeGetByPosition(arguments.back()).column.get();
        if (!level_col->isConst())
            throw Exception{
                "Second argument of function " + getName() + " must be an integral constant",
                ErrorCodes::ILLEGAL_COLUMN
            };

        const auto level = level_col->get64(0);

        const auto col_untyped = block.safeGetByPosition(arguments.front()).column.get();
        if (const auto col_from = typeid_cast<const ColumnString *>(col_untyped))
        {
            const auto size = col_from->size();
            const auto col_to = std::make_shared<ColumnUInt64>(size);
            block.safeGetByPosition(result).column = col_to;

            const auto & chars = col_from->getChars();
            const auto & offsets = col_from->getOffsets();
            auto & out = col_to->getData();

            for (const auto i : ext::range(0, size))
                out[i] = URLHierarchyHashImpl::apply(level,
                    reinterpret_cast<const char *>(&chars[i == 0 ? 0 : offsets[i - 1]]),
                    i == 0 ? offsets[i] - 1 : (offsets[i] - 1 - offsets[i - 1]));
        }
        else if (const auto col_from = typeid_cast<const ColumnConstString *>(col_untyped))
        {
            block.safeGetByPosition(result).column = std::make_shared<ColumnConstUInt64>(
                col_from->size(),
                URLHierarchyHashImpl::apply(level, col_from->getData().data(), col_from->getData().size()));
        }
        else
            throw Exception{
                "Illegal column " + block.safeGetByPosition(arguments[0]).column->getName() +
                " of argument of function " + getName(),
                ErrorCodes::ILLEGAL_COLUMN};
    }
};


struct NameHalfMD5             { static constexpr auto name = "halfMD5"; };
struct NameSipHash64        { static constexpr auto name = "sipHash64"; };
struct NameIntHash32         { static constexpr auto name = "intHash32"; };
struct NameIntHash64         { static constexpr auto name = "intHash64"; };

struct ImplCityHash64
{
    static constexpr auto name = "cityHash64";
    using uint128_t = uint128;

    static auto Hash128to64(const uint128_t & x) { return ::Hash128to64(x); }
    static auto Hash64(const char * const s, const std::size_t len) { return CityHash64(s, len); }
};

struct ImplFarmHash64
{
    static constexpr auto name = "farmHash64";
    using uint128_t = farmhash::uint128_t;

    static auto Hash128to64(const uint128_t & x) { return farmhash::Hash128to64(x); }
    static auto Hash64(const char * const s, const std::size_t len) { return farmhash::Hash64(s, len); }
};

struct ImplMetroHash64
{
    static constexpr auto name = "metroHash64";
    using uint128_t = uint128;

    static auto Hash128to64(const uint128_t & x) { return ::Hash128to64(x); }
    static auto Hash64(const char * const s, const std::size_t len)
    {
        union {
            UInt64 u64;
            UInt8 u8[sizeof(u64)];
        };

        metrohash64_1(reinterpret_cast<const UInt8 *>(s), len, 0, u8);

        return u64;
    }
};

using FunctionHalfMD5 = FunctionStringHash64<HalfMD5Impl, NameHalfMD5>;
using FunctionSipHash64 = FunctionStringHash64<SipHash64Impl, NameSipHash64>;
using FunctionIntHash32 = FunctionIntHash<IntHash32Impl, NameIntHash32>;
using FunctionIntHash64 = FunctionIntHash<IntHash64Impl, NameIntHash64>;
using FunctionMD5 = FunctionStringHashFixedString<MD5Impl>;
using FunctionSHA1 = FunctionStringHashFixedString<SHA1Impl>;
using FunctionSHA224 = FunctionStringHashFixedString<SHA224Impl>;
using FunctionSHA256 = FunctionStringHashFixedString<SHA256Impl>;
using FunctionSipHash128 = FunctionStringHashFixedString<SipHash128Impl>;
using FunctionCityHash64 = FunctionNeighbourhoodHash64<ImplCityHash64>;
using FunctionFarmHash64 = FunctionNeighbourhoodHash64<ImplFarmHash64>;
using FunctionMetroHash64 = FunctionNeighbourhoodHash64<ImplMetroHash64>;

}