DROP VIEW IF EXISTS public.kill_task_for_redis;
CREATE OR REPLACE VIEW public.kill_task_for_redis AS
SELECT now() AS sampletime, * FROM pgxc_parallel_query('all', 'WITH lock_mode_conflicts(lock_mode_1, lock_mode_2) AS MATERIALIZED(
                VALUES
                (''AccessShareLock'', ''AccessExclusiveLock''),
                (''RowShareLock'', ''ExclusiveLock''),
                (''RowShareLock'', ''AccessExclusiveLock''),
                (''RowExclusiveLock'', ''ShareLock''),
                (''RowExclusiveLock'', ''ShareRowExclusiveLock''),
                (''RowExclusiveLock'', ''ExclusiveLock''),
                (''RowExclusiveLock'', ''AccessExclusiveLock''),
                (''ShareUpdateExclusiveLock'', ''ShareUpdateExclusiveLock''),
                (''ShareUpdateExclusiveLock'', ''ShareLock''),
                (''ShareUpdateExclusiveLock'', ''ShareRowExclusiveLock''),
                (''ShareUpdateExclusiveLock'', ''ExclusiveLock''),
                (''ShareUpdateExclusiveLock'', ''AccessExclusiveLock''),
                (''ShareLock'', ''RowExclusiveLock''),
                (''ShareLock'', ''ShareUpdateExclusiveLock''),
                (''ShareLock'', ''ShareRowExclusiveLock''),
                (''ShareLock'', ''ExclusiveLock''),
                (''ShareLock'', ''AccessExclusiveLock''),
                (''ShareLock'',''UpdateExclusiveLock''),
                (''ShareRowExclusiveLock'', ''RowExclusiveLock''),
                (''ShareRowExclusiveLock'', ''ShareUpdateExclusiveLock''),
                (''ShareRowExclusiveLock'', ''ShareLock''),
                (''ShareRowExclusiveLock'', ''ShareRowExclusiveLock''),
                (''ShareRowExclusiveLock'', ''ExclusiveLock''),
                (''ShareRowExclusiveLock'', ''AccessExclusiveLock''),
                (''ShareRowExclusiveLock'',''UpdateExclusiveLock''),
                (''ExclusiveLock'', ''RowShareLock''),
                (''ExclusiveLock'', ''RowExclusiveLock''),
                (''ExclusiveLock'', ''ShareUpdateExclusiveLock''),
                (''ExclusiveLock'', ''ShareLock''),
                (''ExclusiveLock'', ''ShareRowExclusiveLock''),
                (''ExclusiveLock'', ''ExclusiveLock''),
                (''ExclusiveLock'', ''AccessExclusiveLock''),
                (''ExclusiveLock'',''UpdateExclusiveLock''), 
                (''AccessExclusiveLock'', ''AccessShareLock''),
                (''AccessExclusiveLock'', ''RowShareLock''),
                (''AccessExclusiveLock'', ''RowExclusiveLock''),
                (''AccessExclusiveLock'', ''ShareUpdateExclusiveLock''),
                (''AccessExclusiveLock'', ''ShareLock''),
                (''AccessExclusiveLock'', ''ShareRowExclusiveLock''),
                (''AccessExclusiveLock'', ''ExclusiveLock''),
                (''AccessExclusiveLock'', ''AccessExclusiveLock''),
                (''AccessExclusiveLock'', ''UpdateExclusiveLock''),
                (''UpdateExclusiveLock'' ,''ShareLock''), 
                (''UpdateExclusiveLock'' ,''ShareRowExclusiveLock''), 
                (''UpdateExclusiveLock'' ,''ExclusiveLock''), 
                (''UpdateExclusiveLock'' ,''AccessExclusiveLock''), 
                (''UpdateExclusiveLock'' ,''UpdateExclusiveLock'')
            ),
            lock_info AS(
                SELECT /*+ no merge(lc) no tablescan(c) indexscan(p) leading((lc c p)) nestloop(lc c) nestloop(lc c p)  */
                    lc.lockFlag,
                    lc.locktype,
                    lc.stmt_type,
                    pg_catalog.pgxc_node_str() nodename,
                    lc.datname,
                    lc.rolname AS usename,
                    n.nspname,
                    c.relname,
                    p.relname partname,
                    lc.page,
                    lc.tuple,
                    lc.transactionid,
                    lc.mode,
                    lc.granted,
                    lc.client_addr,
                    lc.application_name,
                    lc.pid,
                    CASE WHEN lc.gxid::text <> 0::text THEN lc.gxid ELSE lc.next_xid END gxid,
                    lc.xact_start,
                    lc.query_start,
                    lc.state,
                    lc.query_id,
                    lc.query
                FROM (SELECT 
                        l.*, d.datname,r.rolname,rx.gxid, rx.next_xid,
                        a.stmt_type,a.client_addr, a.application_name, a.xact_start, a.query_start, a.state, a.query_id, a.query,
                    (l.locktype || ''+'' || d.oid || ''+'' || l.relation || ''+'' || l.page || ''+'' || l.tuple || ''+'' || l.transactionid || ''+'' || classid || ''+'' || l.objid) AS lockFlag
                    FROM pg_catalog.pg_locks l
                    INNER JOIN pg_stat_get_activity_with_conninfo(NULL) a ON a.pid = l.pid
                    INNER JOIN pg_database d ON d.oid = a.datid AND d.datname = current_database()
                    INNER JOIN pg_catalog.pg_roles r ON r.oid = a.usesysid
                    INNER JOIN pg_catalog.pg_get_running_xacts() rx ON (a.pid = rx.pid)
                    WHERE 1 = 1
                    AND r.rolname NOT IN(''Ruby'')
                    AND l.locktype IN (''relation'',  ''partition'')
                    AND now() - xact_start > INTERVAL ''{}s''
                ) lc
                LEFT JOIN pg_catalog.pg_class c ON (lc.relation = c.oid OR lc.classid = c.oid) 
                LEFT JOIN pg_catalog.pg_namespace n ON (c.relnamespace = n.oid) 
                LEFT JOIN pg_catalog.pg_partition p ON (lc.classid = p.parentid AND lc.objid = p.oid) 
            )
            SELECT
                nodename,
                locktype,
                datname,
                nspname,
                relname,
                partname,
                waitime,
                acquire_lock_pid,
                hold_lock_pid,
                acquire_lock_event,
                hold_lock_event,
                kill_type
            FROM(
                SELECT /*+ no merge(t1) no merge(t2)  */
                    t1.nodename,
                    t1.locktype,
                    t1.datname,
                    t1.nspname,
                    t1.relname,
                    t1.partname,
                    (now()::timestamp-t1.xact_start::timestamp) AS waitime,
                    t1.pid AS acquire_lock_pid,
                    t2.pid AS hold_lock_pid,
                    (
                        ''usename           : '' || t1.usename  || E''\n'' ||
                        ''mode              : '' || t1.mode || E''\n'' ||
                        ''client_addr       : '' || t1.client_addr || E''\n'' ||
                        ''application_name  : '' || t1.application_name || E''\n'' ||
                        ''gxid              : '' || t1.gxid || E''\n'' ||
                        ''xact_start        : '' || t1.xact_start || E''\n'' ||
                        ''query_start       : '' || t1.query_start || E''\n'' ||
                        ''state             : '' || t1.state || E''\n'' ||
                        ''stmt_type         : '' || t1.stmt_type || E''\n'' ||
                        ''query_id          : '' || t1.query_id || E''\n'' ||
                        ''query             : '' || t1.query || E''\n'' ||
                        ''------------------------------------------------------''
                    ) AS acquire_lock_event,
                    (
                        ''usename           : '' || t2.usename  || E''\n'' ||
                        ''mode              : '' || t2.mode || E''\n'' ||
                        ''client_addr       : '' || t2.client_addr || E''\n'' ||
                        ''application_name  : '' || t2.application_name || E''\n'' ||
                        ''gxid              : '' || t2.gxid || E''\n'' ||
                        ''xact_start        : '' || t2.xact_start || E''\n'' ||
                        ''query_start       : '' || t2.query_start || E''\n'' ||
                        ''state             : '' || t2.state || E''\n'' ||
                        ''stmt_type         : '' || t2.stmt_type || E''\n'' ||
                        ''query_id          : '' || t2.query_id || E''\n'' ||
                        ''query             : '' || t2.query || E''\n'' ||
                        ''------------------------------------------------------''
                    ) AS hold_lock_event,
                    CASE WHEN t1.application_name = ''gs_redis''
                             THEN 1 
                         ELSE 0
                    END event_type,
                    CASE WHEN t2.state = ''idle in transaction''  AND pg_terminate_backend(t2.pid) THEN ''terminate'' -- idle in transaction 执行 terminate
                         WHEN  pg_cancel_backend(t2.pid) THEN ''cancel''     -- 非 idle in transaction 执行 cancel
                    END AS kill_type
                FROM lock_info t1
                JOIN lock_info t2 ON (t1.lockFlag = t2.lockFlag AND t1.granted = false AND t2.granted = true) 
                JOIN lock_mode_conflicts lc ON (t1.mode = lc.lock_mode_1 AND t2.mode = lc.lock_mode_2)
                WHERE event_type > 0
            )
            ') AS(
             nodename            name     ,
             locktype            text     ,
             datname             name     ,
             nspname             name     ,
             relname             name     ,
             partname            name     ,
             waitime             interval ,
             acquire_lock_pid    bigint   ,
             hold_lock_pid       bigint   ,
             acquire_lock_event  text     ,
             hold_lock_event     text     ,
             kill_type           text)
;


DROP TABLE IF EXISTS public.kill_task_history_redis;
CREATE TABLE public.kill_task_history_redis (
                                                sampletime timestamptz,
                                                nodename name,
                                                locktype text,
                                                datname name,
                                                nspname name,
                                                relname name,
                                                partname name,
                                                waitime interval,
                                                acquire_lock_pid bigint,
                                                hold_lock_pid bigint,
                                                acquire_lock_event text,
                                                hold_lock_event text,
                                                kill_type text
)
    WITH (orientation=row, compression=no, period='1 day', TTL= '12 month')
    DISTRIBUTE BY ROUNDROBIN
    PARTITION BY RANGE(sampletime);