# coding=utf-8
#
# 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.

# HITS

# Please refer to the hits.sql_in file for the documentation

"""
@file hits.py_in

@namespace graph
"""

import math
import plpy
import sys
from graph_utils import get_graph_usage
from graph_utils import get_default_threshold_for_link_analysis
from graph_utils import update_output_grouping_tables_for_link_analysis
from graph_utils import validate_graph_coding
from graph_utils import validate_params_for_link_analysis

from utilities.control import MinWarning
from utilities.utilities import _assert
from utilities.utilities import _check_groups
from utilities.utilities import get_table_qualified_col_str
from utilities.utilities import add_postfix
from utilities.utilities import extract_keyvalue_params
from utilities.utilities import unique_string, split_quoted_delimited_str
from utilities.utilities import is_platform_pg

from utilities.validate_args import columns_exist_in_table, drop_tables
from utilities.validate_args import get_cols_and_types, table_exists
from utilities.utilities import rename_table

def validate_hits_args(schema_madlib, vertex_table, vertex_id, edge_table,
                       edge_params, out_table, max_iter, threshold,
                       grouping_cols_list=None):
    """
    Function to validate input parameters for HITS
    """
    validate_graph_coding(vertex_table, vertex_id, edge_table, edge_params,
                          out_table, 'HITS')
    # Validate args such as threshold and max_iter
    validate_params_for_link_analysis(schema_madlib, "HITS",
                                            threshold, max_iter,
                                            edge_table, grouping_cols_list)


def hits(schema_madlib, vertex_table, vertex_id, edge_table, edge_args,
         out_table, max_iter, threshold, grouping_cols, **kwargs):
    """
    Function that computes the HITS scores
    Args:
        @param schema_madlib:
        @param vertex_table:
        @param vertex_id:
        @param edge_table:
        @param edge_args:
        @param out_table:
        @param max_iter:
        @param threshold:
        @param grouping_cols:
        @param kwargs:
    """
    with MinWarning('warning'):
        params_types = {'src': str, 'dest': str}
        default_args = {'src': 'src', 'dest': 'dest'}
        edge_params = extract_keyvalue_params(
            edge_args, params_types, default_args)

        # populate default values for optional params if null
        if max_iter is None:
            max_iter = 100
        if not vertex_id:
            vertex_id = "id"
        if not grouping_cols:
            grouping_cols = ''

        grouping_cols_list = split_quoted_delimited_str(grouping_cols)
        validate_hits_args(schema_madlib, vertex_table, vertex_id, edge_table,
                           edge_params, out_table, max_iter, threshold,
                           grouping_cols_list)

        summary_table = add_postfix(out_table, "_summary")
        _assert(not table_exists(summary_table),
                """Graph HITS: Output summary table ({summary_table}) already
                exists.""".format(**locals()))

        src = edge_params["src"]
        dest = edge_params["dest"]
        n_vertices = plpy.execute("""
                            SELECT COUNT({0}) AS cnt
                            FROM {1}
                        """.format(vertex_id, vertex_table))[0]["cnt"]

        if threshold is None:
            threshold = get_default_threshold_for_link_analysis(n_vertices)

        # table/column names used when grouping_cols is set.
        edge_temp_table = unique_string(desp='temp_edge')
        grouping_cols_comma = grouping_cols + ',' if grouping_cols else ''
        distribution = ('' if is_platform_pg() else
                        "DISTRIBUTED BY ({0}{1})".format(
                            grouping_cols_comma, dest))
        drop_tables([edge_temp_table])
        plpy.execute("""
                CREATE TEMP TABLE {edge_temp_table} AS
                SELECT * FROM {edge_table}
                {distribution}
            """.format(**locals()))

        ######################################################################
        # Set initial authority_norm and hub_norm as 1, so that later the final
        # norm should be positive number
        authority_init_value = 1.0
        hub_init_value = 1.0

        subquery1 = unique_string(desp='subquery1')

        distinct_grp_table = ''
        select_grouping_cols_comma = ''
        select_subquery1_grouping_cols_comma = ''
        group_by_clause = ''
        grouping_cols_for_create_table = ''
        grouping_cols_for_create_table_comma = ''
        # This table is created only when grouping is used.
        temp_summary_table = None
        if grouping_cols:
            distinct_grp_table = unique_string(desp='grp')
            drop_tables([distinct_grp_table])
            plpy.execute("""CREATE TEMP TABLE {distinct_grp_table} AS
                    SELECT DISTINCT {grouping_cols} FROM {edge_temp_table}
                """.format(**locals()))
            group_by_clause = get_table_qualified_col_str(subquery1, grouping_cols_list)
            select_grouping_cols_comma = group_by_clause + ','
            select_subquery1_grouping_cols_comma = grouping_cols + ','
            group_by_clause = 'GROUP BY ' + grouping_cols + ',' + vertex_id
            cols_names_types = get_cols_and_types(edge_table)
            grouping_cols_for_create_table = ', '.join([c_name + " " + c_type
                                                        for (c_name, c_type)
                                                        in cols_names_types
                                                        if c_name in
                                                        grouping_cols_list])
            grouping_cols_for_create_table_comma = \
                grouping_cols_for_create_table + ','
            temp_summary_table = unique_string(desp='temp_summary')
            drop_tables([temp_summary_table])
            plpy.execute("""
                CREATE TEMP TABLE {temp_summary_table} (
                    {grouping_cols_for_create_table}
                )
            """.format(**locals()))
            # Create output table. This will be updated whenever a group converges
            # Note that vertex_id is assumed to be an integer (as described in
            # documentation)
            plpy.execute("""
                    CREATE TABLE {out_table} (
                        {grouping_cols_for_create_table_comma}
                        {vertex_id} BIGINT,
                        authority DOUBLE PRECISION,
                        hub DOUBLE PRECISION
                    )
                """.format(**locals()))

        # Intermediate tables required.
        cur = unique_string(desp='cur')
        message = unique_string(desp='message')
        # curalias and msgalias are used as aliases for current and
        # message tables respectively during self joins
        curalias = unique_string(desp='curalias')
        msgalias = unique_string(desp='msgalias')
        message_unconv = unique_string(desp='message_unconv')
        subquery2 = unique_string(desp='subquery2')

        # GPDB has distributed by clauses to help them with indexing.
        # For Postgres we add the index explicitly.
        if is_platform_pg():
            plpy.execute("CREATE INDEX ON {0}({1})".format(
                edge_temp_table, dest))

        if is_platform_pg():
            cur_distribution = cnts_distribution = ''
        else:
            cur_distribution = cnts_distribution = \
                "DISTRIBUTED BY ({0}{1})".format(
                    grouping_cols_comma, vertex_id)
        cur_join_clause = " {cur}.{vertex_id} = {edge_temp_table}.{dest}"\
            .format(**locals())
        curalias_join_clause = "{curalias}.{vertex_id} = {edge_temp_table}.{src}"\
            .format(**locals())
        drop_tables([cur])
        plpy.execute("""
            CREATE TEMP TABLE {cur} AS
            SELECT {select_grouping_cols_comma} {subquery1}.{vertex_id},
                    {authority_init_value}::DOUBLE PRECISION AS authority,
                    {hub_init_value}::DOUBLE PRECISION AS hub
            FROM (
                SELECT {select_subquery1_grouping_cols_comma} {vertex_table}.{vertex_id}
                FROM {edge_temp_table} JOIN {vertex_table}
                ON {edge_temp_table}.{src}={vertex_table}.{vertex_id}
                UNION
                SELECT {select_subquery1_grouping_cols_comma} {vertex_table}.{vertex_id}
                FROM {edge_temp_table} JOIN {vertex_table}
                ON {edge_temp_table}.{dest}={vertex_table}.{vertex_id}
            ) {subquery1}
            {group_by_clause}
            {cur_distribution}
        """.format(**locals()))

        # The summary table contains the total number of iterations for each
        # group
        plpy.execute("""
                CREATE TABLE {summary_table} (
                    {grouping_cols_for_create_table_comma}
                    __iterations__ INTEGER
                )
            """.format(**locals()))

        # create message table
        cur_grouping_cols_comma = ''
        grouping_cols_join_condition_cur_and_edge = ''
        grouping_cols_join_condition_cur_and_curalias = ''

        # update message table
        msg_grouping_cols_comma = ''
        grouping_cols_where_clause_msg_subquery2 = ''
        grouping_cols_join_condition_msg_and_edge = ''
        grouping_cols_join_condition_msg_and_msgalias = ''
        grouping_cols_where_condition_msg_subquery1 = ''
        group_by_msg_grouping_cols = ''

        # check for convergence
        select_distinct_unconverged_rows = '{0}.{1}'.format(cur, vertex_id)
        grouping_cols_join_condition_cur_and_msg = ''

        if grouping_cols:
            cur_grouping_cols = get_table_qualified_col_str(cur, grouping_cols_list)
            cur_grouping_cols_comma = cur_grouping_cols + ','
            grouping_cols_join_condition_cur_and_edge = ' AND ' + \
                _check_groups(cur, edge_temp_table, grouping_cols_list)
            grouping_cols_join_condition_cur_and_curalias = ' AND ' + \
                _check_groups(cur, curalias, grouping_cols_list)

            msg_grouping_cols = get_table_qualified_col_str(message, grouping_cols_list)
            msg_grouping_cols_comma = msg_grouping_cols + ','
            grouping_cols_where_clause_msg_subquery2 = ' AND ' + \
                _check_groups(message, subquery2, grouping_cols_list)
            grouping_cols_join_condition_msg_and_edge = ' AND ' + \
                _check_groups(message, edge_temp_table, grouping_cols_list)
            grouping_cols_join_condition_msg_and_msgalias = ' AND ' + \
                _check_groups(message, msgalias, grouping_cols_list)

            group_by_msg_grouping_cols = ' GROUP BY ' + msg_grouping_cols

            grouping_cols_where_condition_msg_subquery1 = ' WHERE ' + \
                _check_groups(message, subquery1, grouping_cols_list)

            # this is used for the message_unconv table to find out how many groups
            # have converged
            select_distinct_unconverged_rows = cur_grouping_cols
            grouping_cols_join_condition_cur_and_msg = ' AND ' + \
                _check_groups(cur, message, grouping_cols_list)

        # Variables common to both grouping and non-grouping cases
        message_join_clause = "{message}.{vertex_id} = \
        {edge_temp_table}.{src}".format(**locals())
        msgalias_join_clause = "{msgalias}.{vertex_id} = {edge_temp_table}.{dest}".format(
            **locals())
        sum_norm_square_root = unique_string(desp='sum_norm_sqr_root')
        cur_unconv = unique_string(desp='cur_unconv')

        converged = False
        iteration_num = 0
        """
        We need to calculate the hub and authority scores for each iteration
        and pass the current iteration values to the next iteration.
        To achieve this, we use the following tables
        1. cur -> This gets initialized with default values for both authority
        and hub.
        2. message -> This gets created for each iteration with newly
        computed scores based on cur. At the end of each iteration, we rename
        message to cur as a way to pass authority and hub values to the next
        iteration.
        This convention is similar to message passing paradigm in
        distributed systems such as spark.
        """
        for iteration_num in range(max_iter):

            calculate_authority_and_hub_scores(**locals())

            # Check for convergence only if threshold != 0.
            if threshold != 0:

                converged = check_for_convergence(**locals())

                if iteration_num > 0 and grouping_cols:
                    # Update result and summary tables for groups that have
                    # converged
                    # since the last iteration.
                    update_output_grouping_tables_for_link_analysis(temp_summary_table,
                                                                          iteration_num,
                                                                          summary_table,
                                                                          out_table,
                                                                          message,
                                                                          grouping_cols_list,
                                                                          cur_unconv,
                                                                          message_unconv)

                drop_tables([cur_unconv])
                plpy.execute("""ALTER TABLE {message_unconv} RENAME TO
                    {cur_unconv} """.format(**locals()))

            drop_tables([cur])
            plpy.execute("""ALTER TABLE {message} RENAME TO {cur}
                    """.format(**locals()))
            drop_tables([sum_norm_square_root])

            if converged:
                break

        update_final_results(schema_madlib, converged, threshold, cur, temp_summary_table,
                             iteration_num, summary_table, out_table,
                             grouping_cols_list, cur_unconv, distinct_grp_table)

        # Cleanup All the intermediate tables
        drop_tables([cur, message, cur_unconv, message_unconv, edge_temp_table])

        if grouping_cols:
            drop_tables([distinct_grp_table, temp_summary_table])


def update_final_results(schema_madlib, converged, threshold, cur, temp_summary_table,
                         iteration_num, summary_table, out_table,
                         grouping_cols_list, cur_unconv, distinct_grp_table):
    """
        If there still are some converged/unconverged nodes (within groups/entire
        table), update results table for those nodes.
    """
    if grouping_cols_list:
        if not converged:
            if threshold != 0:
                # We completed max_iters, but there are still some unconverged
                # groups # Update the result and summary tables for unconverged
                # groups.
                update_output_grouping_tables_for_link_analysis(temp_summary_table,
                                                                      iteration_num,
                                                                      summary_table,
                                                                      out_table, cur,
                                                                      grouping_cols_list,
                                                                      cur_unconv)
            else:
                # No group has converged. List of all group values are in
                # distinct_grp_table.
                update_output_grouping_tables_for_link_analysis(temp_summary_table,
                                                                      iteration_num,
                                                                      summary_table,
                                                                      out_table, cur,
                                                                      grouping_cols_list,
                                                                      distinct_grp_table)
    else:

        rename_table(schema_madlib, cur, out_table)
        plpy.execute("""
                INSERT INTO {summary_table} VALUES
                ({iteration_num}+1)
                """.format(**locals()))


def check_for_convergence(**kwargs):
    # message_unconv and cur_unconv will contain the unconverged
    # groups after current and previous iterations respectively.
    # we check if there is at least one unconverged node (limit 1
    # is used in the query).
    plpy.execute("""
                    CREATE TABLE {message_unconv} AS
                    SELECT DISTINCT {select_distinct_unconverged_rows}
                    FROM {message}
                    INNER JOIN {cur}
                    ON {cur}.{vertex_id}={message}.{vertex_id}
                    {grouping_cols_join_condition_cur_and_msg}
                    WHERE ABS({cur}.authority-{message}.authority) > {threshold}
                    OR ABS({cur}.hub-{message}.hub) > {threshold}
                """.format(**kwargs))
    unconverged_node_num = plpy.execute("""
                        SELECT COUNT(*) AS cnt FROM {message_unconv}
                    """.format(**kwargs))[0]["cnt"]
    return unconverged_node_num == 0


def calculate_authority_and_hub_scores(**kwargs):
    """
    This function is responsible for calculating the authority and hub scores.
    This is done in a two-step process:
    1. create message table and compute authority score.
    2. Use authority scores computed to update the hub score.

    :param kwargs: dict of locals() of the calling function.
    :return:
    """
    ###################################################################
    # HITS scores for nodes in a graph at any given iteration 'i' is
    # calculated as following:
    # authority_i(A) = hub_i(B) + hub_i(C) + ..., where B, C are nodes
    # that have edges that point to node A
    # After calculating authority scores for all nodes, hub scores are
    # calculated as following:
    # hub_i(A) = authority_i(D) + authority_i(E) + ..., where D, E are
    # nodes that A points to
    # At the end of each iteration, a normalization will
    # be done for all authority scores and hub scores using L2 distance
    ###################################################################

    ###################################################################
    # calculate authority
    # if there is no node that point to A, authority_i(A) = 0
    ###################################################################
    plpy.execute("""
                    CREATE TABLE {message} AS
                    SELECT {cur_grouping_cols_comma} {cur}.{vertex_id} AS {vertex_id},
                            COALESCE(SUM({curalias}.hub), 0.0) AS authority,
                            {cur}.hub AS hub
                    FROM {cur}
                        LEFT JOIN {edge_temp_table} ON
                        {cur_join_clause} {grouping_cols_join_condition_cur_and_edge}
                        LEFT JOIN {cur} AS {curalias} ON
                        {curalias_join_clause} {grouping_cols_join_condition_cur_and_curalias}
                    GROUP BY {cur_grouping_cols_comma} {cur}.{vertex_id}, {cur}.hub
                    {cur_distribution}
                """.format(**kwargs))
    ###################################################################
    # calculate hub
    # if node A doesn't point to any node, hub_i(A) = 0
    ###################################################################
    plpy.execute("""
                    UPDATE {message}
                    SET hub = {subquery2}.hub
                    FROM
                    (SELECT {msg_grouping_cols_comma} {message}.{vertex_id} AS {vertex_id},
                            COALESCE(SUM({msgalias}.authority), 0) AS hub
                    FROM {message}
                        LEFT JOIN {edge_temp_table} ON
                        {message_join_clause} {grouping_cols_join_condition_msg_and_edge}
                        LEFT JOIN {message} AS {msgalias} ON
                        {msgalias_join_clause} {grouping_cols_join_condition_msg_and_msgalias}
                    GROUP BY {msg_grouping_cols_comma} {message}.{vertex_id}) AS {subquery2}
                    WHERE {subquery2}.{vertex_id} = {message}.{vertex_id}
                    {grouping_cols_where_clause_msg_subquery2}
                """.format(**kwargs))
    # normalize authority and hub score with L2 distance
    plpy.execute("""
                    CREATE TEMP TABLE {sum_norm_square_root} AS
                        SELECT {msg_grouping_cols_comma}
                            SQRT(SUM(POWER(authority, 2))) AS auth_sum_norm_square_root,
                            SQRT(SUM(POWER(hub, 2))) AS hub_sum_norm_square_root
                        FROM {message}
                    {group_by_msg_grouping_cols}
                """.format(**kwargs))

    num_zero_sum_norm_square_root = plpy.execute("""
                                                    SELECT COUNT(*) AS cnt
                                                    FROM {sum_norm_square_root}
                                                    WHERE auth_sum_norm_square_root = 0
                                                    OR hub_sum_norm_square_root = 0
                                                """.format(**kwargs))[0]["cnt"]

    if num_zero_sum_norm_square_root > 0:
        plpy.error("Error while normalizing authority score, please \
                            make sure your graph is a directed graph")
    plpy.execute("""
                    UPDATE {message}
                        SET authority = {message}.authority/{subquery1}.auth_sum_norm_square_root,
                            hub = {message}.hub/{subquery1}.hub_sum_norm_square_root
                    from (SELECT {grouping_cols_comma}
                            auth_sum_norm_square_root,
                            hub_sum_norm_square_root
                        FROM {sum_norm_square_root}) {subquery1}
                    {grouping_cols_where_condition_msg_subquery1}
                """.format(**kwargs))


def hits_help(schema_madlib, message, **kwargs):
    """
    Help function for hits

    Args:
        @param schema_madlib
        @param message: string, Help message string
        @param kwargs

    Returns:
        String. Help/usage information
    """
    if message is not None and \
            message.lower() in ("usage", "help", "?"):
        help_string = "Get from method below"
        help_string = get_graph_usage(schema_madlib, 'HITS',
     """out_table     TEXT, -- Name of the output table for HITS
        max_iter      INTEGER, -- Maximum iteration number (DEFAULT = 100)
        threshold     DOUBLE PRECISION, -- Stopping criteria (DEFAULT = 1/(N*1000),
                                        -- N is number of vertices in the graph)
        grouping_cols TEXT -- Comma separated column names to group on
                           -- (DEFAULT = NULL, no grouping)
""") + """

A summary table is also created that contains information regarding the
number of iterations required for convergence. It is named by adding the
suffix '_summary' to the 'out_table' parameter.
"""
    else:
        help_string = """
----------------------------------------------------------------------------
                                SUMMARY
----------------------------------------------------------------------------
Given a directed graph, hits algorithm finds the authority and hub scores of
all the vertices in the graph.
--
For an overview on usage, run:
SELECT {schema_madlib}.hits('usage');
"""

    return help_string.format(schema_madlib=schema_madlib)
# ---------------------------------------------------------------------