# 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.

# PageRank

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

"""
@file pagerank.py_in

@namespace graph
"""

import plpy
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_output_and_summary_tables
from graph_utils import validate_params_for_link_analysis

from utilities.control import MinWarning
from utilities.control import OptimizerControl
from utilities.utilities import _assert
from utilities.utilities import _check_groups
from utilities.utilities import get_filtered_cols_subquery_str
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.utilities import py_list_to_sql_string

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


def validate_pagerank_args(schema_madlib, vertex_table, vertex_id, edge_table,
                           edge_params, out_table, damping_factor, max_iter,
                           threshold, grouping_cols_list, personalization_vertices):
    """
    Function to validate input parameters for PageRank
    """
    validate_graph_coding(vertex_table, vertex_id, edge_table, edge_params,
                          out_table, 'PageRank')
    # Validate args such as threshold and max_iter
    validate_params_for_link_analysis(schema_madlib, "PageRank",
                                      threshold, max_iter,
                                      edge_table, grouping_cols_list)
    _assert(damping_factor >= 0.0 and damping_factor <= 1.0,
            "PageRank: Invalid damping factor value ({0}), must be between 0 and 1.".
            format(damping_factor))


    # Validate against the given set of nodes for Personalized Page Rank
    if personalization_vertices:
        grouping_cols = get_table_qualified_col_str(
            edge_table, grouping_cols_list)
        group_by_clause = "GROUP BY {0}".format(grouping_cols) \
            if grouping_cols_list else ''
        src = edge_params["src"]
        dest = edge_params["dest"]
        input_personalization_vertices_length = len(personalization_vertices)

        personalization_vertices_str = ','.join([str(i) for i in personalization_vertices])

        # Get a list which has the number of personalization nodes of each group
        vertices_count_list_by_group = plpy.execute("""
                SELECT count(distinct {vertex_id}) AS count
                FROM {vertex_table}
                RIGHT JOIN {edge_table}
                ON ({vertex_table}.{vertex_id} = {edge_table}.{src}
                OR {vertex_table}.{vertex_id} = {edge_table}.{dest})
                AND {vertex_table}.{vertex_id} = ANY(ARRAY[{personalization_vertices_str}])
                {group_by_clause}
            """.format(**locals()))


        # The number of personalization nodes for every group should be equal to
        # the number given by input personalization_vertices list. Otherwise,
        # some nodes are missing for certain group. Or there might be duplicate
        # nodes in input personalization_vertices list. Or there are some
        # invalid nodes in input list that don't exist in vertex table. In any
        # case, throw an error.
        for key in vertices_count_list_by_group:
            if key["count"] != input_personalization_vertices_length:
                plpy.error("Personalization nodes must be a subset "
                           "of the vertex_table without duplicates and "
                           "every nodes should be present in all the groups")


def pagerank(schema_madlib, vertex_table, vertex_id, edge_table, edge_args, out_table,
             damping_factor, max_iter, threshold, grouping_cols, personalization_vertices, **kwargs):
    """
    Function that computes the PageRank

    Args:
        @param vertex_table
        @param vertex_id
        @param edge_table
        @param source_vertex
        @param dest_vertex
        @param out_table
        @param damping_factor
        @param max_iter
        @param threshold
        @param personalization_vertices
    """
    # We noticed the current implementation is noticeably slower when orca=ON
    # on Greenplum 5, so that we temporarily turn if off to run pagerank.
    with OptimizerControl(False):

        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 damping_factor is None:
                damping_factor = 0.85
            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_pagerank_args(schema_madlib, vertex_table, vertex_id, edge_table,
                                   edge_params, out_table, damping_factor,
                                   max_iter, threshold, grouping_cols_list,
                                   personalization_vertices)

            summary_table = add_postfix(out_table, "_summary")
            _assert(not table_exists(summary_table),
                    "Graph PageRank: 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.
            distinct_grp_table = ''
            vertices_per_group = ''
            vpg = ''
            grouping_where_clause = ''
            group_by_clause = ''
            random_prob = ''
            ppr_join_clause = ''

            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))
            plpy.execute("DROP TABLE IF EXISTS {0}".format(edge_temp_table))
            plpy.execute("""
                CREATE TEMP TABLE {edge_temp_table} AS
                    SELECT * FROM {edge_table}
                    {distribution}
                """.format(**locals()))

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

            # Intermediate tables required.
            cur = unique_string(desp='cur')
            message = unique_string(desp='message')
            cur_unconv = unique_string(desp='cur_unconv')
            message_unconv = unique_string(desp='message_unconv')
            out_cnts = unique_string(desp='out_cnts')
            out_cnts_cnt = unique_string(desp='cnt')
            v1 = unique_string(desp='v1')

            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 = """{edge_temp_table}.{dest} = {cur}.{vertex_id}
                """.format(**locals())
            out_cnts_join_clause = """{out_cnts}.{vertex_id} =
                {edge_temp_table}.{src} """.format(**locals())
            v1_join_clause = """{v1}.{vertex_id} = {edge_temp_table}.{src}
                """.format(**locals())

            # Get query params for Personalized Page Rank.
            ppr_params = ''
            total_ppr_nodes = 0
            random_jump_prob_ppr = 0
            ppr_init_value_clause = ''
            if personalization_vertices:
                ppr_params = get_query_params_for_ppr(personalization_vertices, damping_factor,
                                                      vertex_id, edge_temp_table, vertex_table, edge_params)
                total_ppr_nodes = ppr_params[0]
                random_jump_prob_ppr = ppr_params[1]
                ppr_init_value_clause = ppr_params[2]

            random_probability = (1.0 - damping_factor) / n_vertices
            if total_ppr_nodes > 0:
                random_jump_prob = random_jump_prob_ppr
            else:
                random_jump_prob = random_probability

            ####################################################################
            # Create several strings that will be used to construct required
            # queries. These strings will be required only during grouping.

            ignore_group_clause_first = ''
            limit = ' LIMIT 1 '

            grouping_cols_select_pr = ''
            vertices_per_group_inner_join_pr = ''
            ignore_group_clause_pr = ''

            grouping_cols_select_ins = ''
            vpg_from_clause_ins = ''
            vpg_where_clause_ins = ''
            message_grp_where_ins = ''
            ignore_group_clause_ins = ''

            nodes_with_no_incoming_edges = unique_string(desp='no_incoming')
            ignore_group_clause_ins_noincoming = ''

            grouping_cols_select_conv = '{0}.{1}'.format(cur, vertex_id)
            group_by_grouping_cols_conv = ''
            message_grp_clause_conv = ''
            ignore_group_clause_conv = ''

            ####################################################################

            # Queries when groups are involved need a lot more conditions in
            # various clauses, so populating the required variables. Some intermediate
            # tables are unnecessary when no grouping is involved, so create some
            # tables and certain columns only during grouping.
            if grouping_cols:
                distinct_grp_table = unique_string(desp='grp')
                plpy.execute(
                    "DROP TABLE IF EXISTS {0}".format(distinct_grp_table))
                plpy.execute("""CREATE TEMP TABLE {distinct_grp_table} AS
                        SELECT DISTINCT {grouping_cols} FROM {edge_temp_table}
                    """.format(**locals()))
                vertices_per_group = unique_string(desp='nvert_grp')
                init_pr = unique_string(desp='init')
                random_prob = unique_string(desp='rand')
                subq = unique_string(desp='subquery')
                rand_damp = 1 - damping_factor
                grouping_where_clause = _check_groups(
                    distinct_grp_table, subq, grouping_cols_list)
                group_by_clause = get_table_qualified_col_str(
                    distinct_grp_table, grouping_cols_list)
                # Find number of vertices in each group, this is the normalizing
                # factor for computing the random_prob
                where_clause_ppr = ''
                if personalization_vertices:
                    personalization_vertices_str = ','.join(
                        [str(i) for i in personalization_vertices])
                    where_clause_ppr = """
                        where __vertices__ = ANY(ARRAY[{personalization_vertices_str}])
                    """.format(**locals())
                    random_prob_grp = 1.0 - damping_factor
                    init_prob_grp = 1.0 / total_ppr_nodes
                else:
                    personalization_vertices_str = ''
                    random_prob_grp = """
                            {rand_damp}/COUNT(__vertices__)::DOUBLE PRECISION
                        """.format(**locals())
                    init_prob_grp = """
                            1/COUNT(__vertices__)::DOUBLE PRECISION
                        """.format(**locals())

                plpy.execute(
                    "DROP TABLE IF EXISTS {0}".format(vertices_per_group))
                plpy.execute("""CREATE TEMP TABLE {vertices_per_group} AS
                        SELECT {distinct_grp_table}.*,
                        {init_prob_grp} AS {init_pr},
                        {random_prob_grp} as {random_prob}
                        FROM {distinct_grp_table} INNER JOIN (
                            SELECT {grouping_cols}, {src} AS __vertices__
                            FROM {edge_temp_table}
                            UNION
                            SELECT {grouping_cols}, {dest} FROM {edge_temp_table}
                        ){subq}
                        ON {grouping_where_clause}
                        {where_clause_ppr}
                        GROUP BY {group_by_clause}
                    """.format(**locals()))

                grouping_where_clause = _check_groups(
                    vertices_per_group, subq, grouping_cols_list)
                group_by_clause = get_table_qualified_col_str(
                    vertices_per_group, grouping_cols_list)

                if personalization_vertices:
                    init_prob_grp_ppr = 1.0 / total_ppr_nodes
                    init_pr = """
                            CASE when __vertices__ = ANY(ARRAY[{personalization_vertices_str}])
                            THEN {init_prob_grp_ppr} ELSE 0 END
                        """.format(**locals())

                plpy.execute("""
                        CREATE TEMP TABLE {cur} AS
                        SELECT {group_by_clause}, {subq}.__vertices__
                                AS {vertex_id}, {init_pr} AS pagerank
                        FROM {vertices_per_group} INNER JOIN (
                            SELECT {grouping_cols}, {src} AS __vertices__
                            FROM {edge_temp_table}
                            UNION
                            SELECT {grouping_cols}, {dest} FROM {edge_temp_table}
                        ){subq}
                        ON {grouping_where_clause}
                        {cur_distribution}
                    """.format(**locals()))
                vpg = unique_string(desp='vpg')

                # Compute the out-degree of every node in the group-based
                # subgraphs.
                plpy.execute("DROP TABLE IF EXISTS {0}".format(out_cnts))
                plpy.execute("""CREATE TEMP TABLE {out_cnts} AS
                    SELECT {grouping_cols_select} {src} AS {vertex_id},
                           COUNT({dest}) AS {out_cnts_cnt}
                    FROM {edge_temp_table}
                    GROUP BY {grouping_cols_select} {src}
                    {cnts_distribution}
                    """.format(grouping_cols_select=grouping_cols + ','
                               if grouping_cols else '', **locals()))

                message_grp = _check_groups(cur, message, grouping_cols_list)
                cur_join_clause = cur_join_clause + ' AND ' + _check_groups(
                    cur, edge_temp_table, grouping_cols_list)

                out_cnts_join_clause = out_cnts_join_clause + ' AND ' \
                    + _check_groups(out_cnts, edge_temp_table,
                                    grouping_cols_list)

                v1_join_clause = v1_join_clause + ' AND ' + _check_groups(
                    v1, edge_temp_table, grouping_cols_list)

                vpg_join_clause = _check_groups(
                    vpg, edge_temp_table, grouping_cols_list)
                vpg_t1_join_clause = _check_groups(
                    vpg, '__t1__', grouping_cols_list)

                # join clause specific to populating random_prob for nodes without any
                # incoming edges.
                edge_grouping_cols_select = get_table_qualified_col_str(
                    edge_temp_table, grouping_cols_list)
                cur_grouping_cols_select = get_table_qualified_col_str(
                    cur, grouping_cols_list)

                # Create output summary table:
                cols_names_types = get_cols_and_types(edge_table)
                grouping_cols_clause = ', '.join([c_name + " " + c_type
                                                  for (c_name, c_type)
                                                  in cols_names_types
                                                  if c_name in grouping_cols_list])
                plpy.execute("""
                        CREATE TABLE {summary_table} (
                            {grouping_cols_clause},
                            __iterations__ INTEGER
                        )
                    """.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_clause},
                            {vertex_id} BIGINT,
                            pagerank DOUBLE PRECISION
                        )
                    """.format(**locals()))
                temp_summary_table = unique_string(desp='temp_summary')
                plpy.execute(
                    "DROP TABLE IF EXISTS {0}".format(temp_summary_table))
                plpy.execute("""
                        CREATE TABLE {temp_summary_table} (
                            {grouping_cols_clause}
                        )
                    """.format(**locals()))
                ################################################################
                # Strings required for the main PageRank computation query
                grouping_cols_select_pr = edge_grouping_cols_select + ', '

                if personalization_vertices:
                    random_jump_prob = random_jump_prob_ppr
                else:
                    random_jump_prob = 'MIN({vpg}.{random_prob})'.format(
                        **locals())
                vertices_per_group_inner_join_pr = """
                    INNER JOIN {vertices_per_group}
                    AS {vpg} ON {vpg_join_clause}""".format(**locals())

                ignore_group_clause_pr = ' WHERE ' + \
                    get_filtered_cols_subquery_str(edge_temp_table,
                                                   summary_table,
                                                   grouping_cols_list)

                ignore_group_clause_ins_noincoming = ' WHERE ' + \
                    get_filtered_cols_subquery_str(nodes_with_no_incoming_edges,
                                                   summary_table,
                                                   grouping_cols_list)
                # Strings required for updating PageRank scores of vertices that have
                # no incoming edges
                grouping_cols_select_ins = cur_grouping_cols_select + ','
                vpg_from_clause_ins = ', {vertices_per_group} AS {vpg}'.format(
                    **locals())
                vpg_where_clause_ins = ' AND {vpg_t1_join_clause} '.format(
                    **locals())
                message_grp_where_ins = 'WHERE {message_grp}'.format(**locals())
                ignore_group_clause_ins = ' AND ' + get_filtered_cols_subquery_str(
                    cur, summary_table, grouping_cols_list)
                # Strings required for convergence test query
                grouping_cols_select_conv = cur_grouping_cols_select
                group_by_grouping_cols_conv = ' GROUP BY {0}'.format(
                    cur_grouping_cols_select)
                message_grp_clause_conv = '{0} AND '.format(message_grp)
                ignore_group_clause_conv = ' AND ' + get_filtered_cols_subquery_str(
                    cur, summary_table, grouping_cols_list)
                limit = ''

                # Find all nodes, in each group, that have no incoming edges. The PageRank
                # value of these nodes are not updated using the first query in the
                # following for loop. They must be explicitly plugged back in to the
                # message table, with their corresponding group's random_prob as their
                # PageRank values.
                plpy.execute("""
                        CREATE TABLE {nodes_with_no_incoming_edges} AS
                        SELECT {select_group_cols}, __t1__.{src} AS {vertex_id},
                                {vpg}.{random_prob} AS pagerank
                        FROM {edge_temp_table} AS __t1__ {vpg_from_clause_ins}
                        WHERE NOT EXISTS (
                            SELECT 1
                            FROM {edge_temp_table} AS __t2__
                            WHERE __t1__.{src}=__t2__.{dest}
                            AND {where_group_clause}
                        ) {vpg_where_clause_ins}
                        GROUP BY {select_group_cols}, __t1__.{src}, pagerank
                    """.format(
                    select_group_cols=get_table_qualified_col_str(
                        '__t1__', grouping_cols_list),
                    where_group_clause=_check_groups(
                        '__t1__', '__t2__', grouping_cols_list),
                    **locals()))

            else:
                # cur and out_cnts tables can be simpler when no grouping is
                # involved.
                if total_ppr_nodes > 0:
                    init_value = ppr_init_value_clause
                else:
                    init_value = 1.0 / n_vertices
                plpy.execute("""
                        CREATE TEMP TABLE {cur} AS
                        SELECT {vertex_id}, {init_value}::DOUBLE PRECISION AS pagerank
                        FROM {vertex_table}
                        {cur_distribution}
                    """.format(**locals()))

                # Compute the out-degree of every node in the graph.
                plpy.execute("DROP TABLE IF EXISTS {0}".format(out_cnts))
                plpy.execute("""CREATE TEMP TABLE {out_cnts} AS
                    SELECT {src} AS {vertex_id}, COUNT({dest}) AS {out_cnts_cnt}
                    FROM {edge_temp_table}
                    GROUP BY {src}
                    {cnts_distribution}
                    """.format(**locals()))
                # The summary table when there is no grouping will contain only
                # the iteration column. We don't need to create the out_table
                # when no grouping is used since the 'cur' table will be renamed
                # to out_table after pagerank computation is completed.
                plpy.execute("""
                        CREATE TABLE {summary_table} (
                            __iterations__ INTEGER
                        )
                    """.format(**locals()))
                # Find all nodes in the graph that don't have any incoming edges and
                # assign random_prob as their pagerank values.
                plpy.execute("""
                        CREATE TABLE {nodes_with_no_incoming_edges} AS
                        SELECT DISTINCT({src}), {random_probability} AS pagerank
                        FROM {edge_temp_table}
                        EXCEPT
                            (SELECT DISTINCT({dest}), {random_probability} AS pagerank
                            FROM {edge_temp_table})
                    """.format(**locals()))
            unconverged = 0
            iteration_num = 0

            for iteration_num in range(max_iter):
                ################################################################
                # PageRank for node 'A' at any given iteration 'i' is given by:
                # PR_i(A) = damping_factor(PR_i-1(B)/degree(B) +
                #           PR_i-1(C)/degree(C) + ...) + (1-damping_factor)/N
                # where 'N' is the number of vertices in the graph,
                # B, C are nodes that have edges to node A, and
                # degree(node) represents the number of outgoing edges from 'node'
                ################################################################
                # Essentially, the pagerank for a node is based on an aggregate of a
                # fraction of the pagerank values of all the nodes that have incoming
                # edges to it, along with a small random probability.
                # More information can be found at:
                # https://en.wikipedia.org/wiki/PageRank#Damping_factor

                # The query below computes the PageRank of each node using the above
                # formula. A small explanatory note on ignore_group_clause:
                # This is used only when grouping is set. This essentially will have
                # the condition that will help skip the PageRank computation on groups
                # that have converged.

                plpy.execute("""
                        CREATE TABLE {message} AS
                        SELECT {grouping_cols_select_pr}
                                {edge_temp_table}.{dest} AS {vertex_id},
                                SUM(({v1}.pagerank)/{out_cnts}.{out_cnts_cnt})*{damping_factor}+
                                {random_jump_prob} AS  pagerank
                        FROM {edge_temp_table}
                            INNER JOIN {cur} ON {cur_join_clause}
                            INNER JOIN {out_cnts} ON {out_cnts_join_clause}
                            INNER JOIN {cur} AS {v1} ON {v1_join_clause}
                            {vertices_per_group_inner_join_pr}
                        {ignore_group_clause}
                        GROUP BY {grouping_cols_select_pr} {edge_temp_table}.{dest}
                        {cur_distribution}
                    """.format(ignore_group_clause=ignore_group_clause_pr
                               if iteration_num > 0 else ignore_group_clause_first,
                               **locals()))

                # If there are nodes that have no incoming edges, they are not
                # captured in the message table. Insert entries for such nodes,
                # with random_prob.
                plpy.execute("""
                        INSERT INTO {message}
                        SELECT *
                        FROM {nodes_with_no_incoming_edges}
                        {ignore_group_clause}
                    """.format(ignore_group_clause=ignore_group_clause_ins_noincoming
                               if iteration_num > 0 else ignore_group_clause_first,
                               **locals()))
                # Check for convergence:
                # Check for convergence only if threshold != 0.
                if threshold != 0:
                    # message_unconv and cur_unconv will contain the unconverged groups
                    # after current # and previous iterations respectively. Groups that
                    # are missing in message_unconv but appear in cur_unconv are the
                    # groups that have converged after this iteration's computations.
                    # If no grouping columns are specified, then we check if there is
                    # at least one unconverged node (limit 1 is used in the
                    # query).
                    if iteration_num == 0 and grouping_cols:
                        # Hack to address corner case:
                        # With grouping, if there was a graph that converged in
                        # the very first iteration (a complete graph is an eg.
                        # of such a graph), then the pagerank scores for that
                        # group was not showing up in the output. The following
                        # code just prevents convergence in the first iteration.
                        plpy.execute("""
                            CREATE TEMP TABLE {message_unconv} AS
                            SELECT * FROM {distinct_grp_table}
                        """.format(**locals()))
                    else:
                        plpy.execute("""
                            CREATE TEMP TABLE {message_unconv} AS
                            SELECT {grouping_cols_select_conv}
                            FROM {message}
                            INNER JOIN {cur}
                            ON {cur}.{vertex_id}={message}.{vertex_id}
                            WHERE {message_grp_clause_conv}
                                ABS({cur}.pagerank-{message}.pagerank) > {threshold}
                            {ignore_group_clause}
                            {group_by_grouping_cols_conv}
                            {limit}
                        """.format(ignore_group_clause=ignore_group_clause_ins
                                   if iteration_num > 0 else ignore_group_clause_conv,
                                   **locals()))

                    unconverged = plpy.execute("""SELECT COUNT(*) AS cnt FROM {0}
                        """.format(message_unconv))[0]["cnt"]
                    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)

                    plpy.execute("DROP TABLE IF EXISTS {0}".format(cur_unconv))
                    plpy.execute("""ALTER TABLE {message_unconv} RENAME TO
                        {cur_unconv} """.format(**locals()))
                else:
                    # Do not run convergence test if threshold=0, since that implies
                    # the user wants to run max_iter iterations.
                    unconverged = 1

                plpy.execute("DROP TABLE IF EXISTS {0}".format(cur))
                plpy.execute("""ALTER TABLE {message} RENAME TO {cur}
                        """.format(**locals()))
                if unconverged == 0:
                    break

            # If there still are some unconverged groups/(entire table),
            # update results.
            if grouping_cols:
                if unconverged > 0:
                    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)

                # updating the calculated pagerank value in case of
                # Personalized Page Rank.
                # Ref :
                # https://docs.oracle.com/cd/E56133_01/latest/reference/algorithms/pagerank.html
                if total_ppr_nodes > 1:
                    plpy.execute("""UPDATE {out_table} set pagerank =
                                    pagerank / {total_ppr_nodes}::DOUBLE PRECISION
                                    """.format(out_table=out_table,
                                               total_ppr_nodes=total_ppr_nodes))
            else:
                # updating the calculated pagerank value in case of
                # Personalized Page Rank.
                # Ref :
                # https://docs.oracle.com/cd/E56133_01/latest/reference/algorithms/pagerank.html
                if total_ppr_nodes > 1:
                    plpy.execute("""UPDATE {table_name} set pagerank =
                                    pagerank / {total_ppr_nodes}::DOUBLE PRECISION
                                 """.format(table_name=cur,
                                            total_ppr_nodes=total_ppr_nodes))

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

            # Step 4: Cleanup
            plpy.execute("""DROP TABLE IF EXISTS {0},{1},{2},{3},{4},{5},{6}
                """.format(out_cnts, edge_temp_table, cur, message, cur_unconv,
                           message_unconv, nodes_with_no_incoming_edges))
            if grouping_cols:
                plpy.execute("""DROP TABLE IF EXISTS {0},{1},{2}
                    """.format(vertices_per_group, temp_summary_table,
                               distinct_grp_table))


def get_query_params_for_ppr(personalization_vertices, damping_factor,
                             vertex_id, edge_temp_table, vertex_table, edge_params):
    """
     This function will prepare the Join Clause and the condition to Calculate
     the Personalized Page Rank and Returns Total number of user provided
     personalization vertices, A join Clause and a clause to be added to existing
     formula to calculate pagerank.

     Args:
         @param personalization_vertices
         @param damping_factor
         @param vertex_id
         @param edge_temp_table
         @param vertex_table

     Returns :
             (Integer, String, String)

    """
    total_ppr_nodes = 0
    ppr_random_prob_clause = ''
    ppr_init_prob_clause = ''

    if personalization_vertices:
        total_ppr_nodes = len(personalization_vertices)
        ppr_init_value = 1.0 / total_ppr_nodes
        prob_value = 1.0 - damping_factor
        dest = edge_params["dest"]

        personalization_vertices_str = ','.join([str(i) for i in personalization_vertices])

        # In case of PPR, Assign the Random jump probability to the personalization_vertices only.
        # For rest of the nodes, Random jump probability  will be zero.
        ppr_random_prob_clause = """
                CASE WHEN {edge_temp_table}.{dest} = ANY(ARRAY[{personalization_vertices_str}])
                THEN {prob_value} ELSE 0 END
            """.format(**locals())

        ppr_init_prob_clause = """
                CASE WHEN {vertex_id} = ANY(ARRAY[{personalization_vertices_str}])
                THEN {ppr_init_value} ELSE 0 END
            """.format(**locals())
    return(total_ppr_nodes, ppr_random_prob_clause, ppr_init_prob_clause)


def pagerank_help(schema_madlib, message, **kwargs):
    """
    Help function for pagerank

    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, 'PageRank',
                                      """out_table     TEXT, -- Name of the output table for PageRank
        damping_factor DOUBLE PRECISION, -- Damping factor in random surfer model
                                         -- (DEFAULT = 0.85)
        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_col  TEXT,              -- Comma separated column names to group on
                                         -- (DEFAULT = NULL, no grouping)
        personalization_vertices ARRAY OF BIGINT, -- A comma seperated list of vertices
                                                      or nodes for personalized page rank.
""") + """

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, pagerank algorithm finds the PageRank score of all
the vertices in the graph.
--
For an overview on usage, run:
SELECT {schema_madlib}.pagerank('usage');
"""

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