--ACTIVITY_HIST.HISTORY_PG_STAT_ACTIVITY DROP TABLE IF EXISTS activity_hist.history_pg_stat_activity; CREATE TABLE activity_hist.history_pg_stat_activity (   timepoint timestamp without time zone ,   datid             oid  ,    datname           name ,   pid               integer,   usesysid          oid    ,   usename           name   ,   application_name  text   ,   client_addr       inet   ,   client_hostname   text   ,   client_port       integer,   backend_start     timestamp without time zone ,   xact_start        timestamp without time zone ,   query_start       timestamp without time zone ,   state_change      timestamp without time zone ,   wait_event_type   text ,                        wait_event        text ,                      state             text ,                     backend_xid       xid  ,                    backend_xmin      xid  ,                   query             text ,                  backend_type      text ,   queryid           bigint );

CREATE TABLE pg_stat_db_queries (   database_id integer ,   queryid bigint ,   query text ,   max_time double precision );(

CREATE MATERIALIZED VIEW public.mvw_pg_stat_queries AS  SELECT t.queryid,     t.max_time,     t.query    FROM public.dblink('LINK1'::text, 'SELECT queryid , max_time , query FROM pg_stat_statements WHERE dbid=(SELECT oid FROM  pg_database WHERE datname=current_database() ) AND max_time >= 0 '::text) t(queryid bigint, max_time double precision, query text)   WITH NO DATA; 

CREATE TABLE log_query (   id  integer ,   queryid  bigint ,   query_md5hash  text ,   database_id  integer ,   timepoint  timestamp without time zone ,    query  text ,   explained_plan  text[] ,    plan_md5hash  text ,   explained_plan_wo_costs  text[] ,    plan_hash_value  text ,   ip  text,   port  text ,    pid  integer  );

CREATE OR REPLACE FUNCTION refresh_pg_stat_queries_list( database_id int) RETURNS BOOLEAN AS $$ DECLARE  result BOOLEAN ;  database_rec record ;   BEGIN      SELECT *   INTO database_rec   FROM endpoint e JOIN database d ON e.id = d.endpoint_id    WHERE d.id = database_id  ;        IF NOT database_rec.is_need_monitoring THEN RAISE NOTICE 'NO NEED MONITORING FOR database_id=%',database_id; return TRUE ; END IF ;      EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' port=5432 dbname='||database_rec.name|| 		                                         ' user='||database_rec.s_name||' password='||database_rec.s_pass|| ' '')';       REFRESH MATERIALIZED VIEW mvw_pg_stat_queries ;      PERFORM dblink_disconnect('LINK1');      RETURN result; END $$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION refresh_pg_stat_db_queries( ) RETURNS BOOLEAN AS $$ DECLARE  result BOOLEAN ;  database_rec record ;    pg_stat_rec record ; BEGIN    TRUNCATE pg_stat_db_queries;         FOR database_rec IN   SELECT *   FROM database d    LOOP        IF NOT database_rec.is_need_monitoring THEN RAISE NOTICE 'NO NEED MONITORING FOR database_id=%',database_rec.id; CONTINUE ; END IF ;         PERFORM refresh_pg_stat_queries_list( database_rec.id ) ;  	 	FOR pg_stat_rec IN 	SELECT *  	FROM mvw_pg_stat_queries  	LOOP 	  INSERT INTO pg_stat_db_queries 	  ( database_id , queryid , query , max_time ) 	  VALUES 	  ( database_rec.id , pg_stat_rec.queryid , pg_stat_rec.query , pg_stat_rec.max_time); 	END LOOP;        END LOOP;     RETURN TRUE; END $$ LANGUAGE plpgsql;

#!/bin/bash ######################################################### # upload_log_query.sh # Upload table table from dowloaded aws file  # version 12.0 ###########################################################   echo 'TIMESTAMP:'$(date +%c)' Upload log_query table '  source_file=$1 echo 'source_file='$source_file  database_id=$2 echo 'database_id='$database_id  database_name=$3 echo 'database_name='$database_name   beginer=' ' first_line='1' let "line_count=0" sql_line=' ' sql_flag=' '     space=' ' cat $source_file | while read line do   #first line will be passed   if [[ $line_count == '0' ]]; then      let "line_count++"      continue    fi      line="$space$line"   #echo 'line='$line    if [[ $first_line == "1" ]]; then     beginer=`echo $line | awk -F" " '{ print $1}' `     first_line='0'   fi    current_beginer=`echo $line | awk -F" " '{ print $1}' `   #echo 'current_beginer='$current_beginer   #echo 'beginer='$beginer       if [[ $current_beginer == $beginer ]]; then    if [[ $sql_flag == '1' ]]; then      sql_flag='0'       #echo 'TIMESTAMP:'$(date +%c)' Upload log_query table : SQL STATEMENT ='"$sql_line"       log_date=`echo $sql_line | awk -F" " '{ print $1}' `      #echo 'log_date='$log_date       log_time=`echo $sql_line | awk -F" " '{ print $2}' `      #echo 'log_time='$log_time            duration=`echo $sql_line | awk -F" " '{ print $5}' `      #echo 'duration='$duration  	 connect=`echo $sql_line | awk -F" " '{ print $3}' ` 	 userdb=`echo $connect | awk -F":" '{ print $3}' ` 	 userdb2=$userdb'@' 	 db_port_log=`echo $connect | awk -F"@" '{ print $2}' ` 	 log_database_name=`echo $db_port_log | awk -F":" '{ print $1}' ` 	  	  	 #echo 'connect='$connect 	 #echo 'userdb='$userdb 	 #echo 'userdb2='$userdb2 	 #echo 'db_port_log='$db_port_log 	 #echo 'log_database_name='$log_database_name 	  	 if [[  "$log_database_name" != "$database_name" ]]; 	 then 	   echo '*** database_name '$log_database_name' from log is not equal '$database_name' CONTINUE ' 	   continue; 	 fi 	       #replace ' to ''      sql_modline=`echo "$sql_line" | sed 's/'\''/'\'''\''/g'`      sql_line=' ' 	 #echo '*********************************log_query start'      #echo 'pid_str='$pid_str  	 #echo 'ip_port='$ip_port  	 #echo 'database_id='$database_id  	 #echo 'log_date='$log_date 	 #echo 'log_time='$log_time 	 #echo 'duration='$duration 	 #echo 'sql_modline='$sql_modline      if ! psql -U monitor -d monitor -v ON_ERROR_STOP=1 -A -t -q -c "select log_query( '$pid_str' , '$ip_port' , $database_id , '$log_date' , '$log_time' , '$duration' , '$sql_modline' )"      then         echo 'FATAL_ERROR - log_query '         exit 1      fi 	 #echo '**********************************log_query finish'      fi #if [[ $sql_flag == '1' ]]; then      let "line_count=line_count+1"     #echo 'line_count= '$line_count     #echo $line      #check=`echo $line | awk -F" " '{ print $8}' `     #check_sql=${check^^}          #echo 'check_sql='$check_sql      	if [[ ${line^^} =~ "SELECT" ]];  	then  	 if [[ $line =~ "duration:" ]]; 	 then 	    test_statement=`echo $line | awk -F" " '{ print $8}'`  		is_select=${test_statement^^}		 		 		#echo 'test_statement='$test_statement 		#echo 'is_select='$is_select 		         if [[ $is_select == 'SELECT' ]];          then		 		  sql_flag='1'     		  sql_line="$sql_line$line" 		  ip_port=`echo $sql_line | awk -F":" '{ print $4}' ` 		  pid_str=`echo $sql_line | awk -F":" '{ print $6}' ` 		fi 	 fi	      fi   else            #echo $line     #echo 'sql_flag ='$sql_flag      if [[ $sql_flag == '1' ]]; then       sql_line="$sql_line$line"     fi           fi #if [[ $current_beginer == $beginer ]]; then  done

--log_query.sql --insert new query into log_query table CREATE OR REPLACE FUNCTION log_query( pid_str text , ip_port text ,log_database_id integer , log_date text , log_time text , duration text , sql_line text   ) RETURNS boolean AS $$ DECLARE   result boolean ;   log_timepoint timestamp without time zone ;   log_duration double precision ;    pos integer ;   log_query text ;   activity_string text ;   log_md5hash text ;   log_explain_plan text[] ;      log_planhash text ;   log_plan_wo_costs text[] ;       database_rec record ;      pg_stat_query text ;    test_log_query text ;   metric_rec record;   log_query_rec record;   found_flag boolean;      pg_stat_history_rec record ;   port_start integer ;   port_end integer ;   client_ip text ;   client_port text ;   log_queryid bigint ;   log_query_text text ;   pg_stat_query_text text ;    current_pid_str text ;   current_pid integer;   pid_start_pos integer ;   pid_finish_pos integer ;    BEGIN   result = TRUE ;          IF ip_port != '[local]' THEN     port_start = position('(' in ip_port);     port_end = position(')' in ip_port);     client_ip = substring( ip_port from 1 for port_start-1 );     client_port = substring( ip_port from port_start+1 for port_end-port_start-1 );   ELSE      client_ip = 'local'; 	client_port = 'local';   END IF;       pid_start_pos = position('[' in pid_str);   pid_finish_pos = position(']' in pid_str);   current_pid_str=substring( pid_str from 2 for pid_finish_pos - pid_start_pos -1 );   current_pid = to_number(current_pid_str , '999999999999');      SELECT e.host , d.name , d.owner_pwd , d.owner_user   INTO database_rec   FROM database d JOIN endpoint e ON e.id = d.endpoint_id   WHERE d.id = log_database_id ;      log_timepoint = to_timestamp(log_date||' '||log_time,'YYYY-MM-DD HH24-MI-SS');   log_duration = duration::double precision ;         pos = position ('SELECT' in UPPER(sql_line) );   log_query = substring( sql_line from pos for LENGTH(sql_line));   log_query = regexp_replace(log_query,' +',' ','g');   log_query = regexp_replace(log_query,';+','','g');   log_query = trim(trailing ' ' from log_query);      log_md5hash = md5( log_query::text );      --Explain execution plan-- EXECUTE 'SELECT dblink_connect(''LINK1'',''host='||database_rec.host||' port=5432 dbname='||database_rec.name||' user='||database_rec.owner_user||' password='||database_rec.owner_pwd||' '')';       log_explain_plan = ARRAY ( SELECT * FROM dblink('LINK1', 'EXPLAIN '||log_query ) AS t (plan text) );   log_plan_wo_costs = ARRAY ( SELECT * FROM dblink('LINK1', 'EXPLAIN ( COSTS FALSE ) '||log_query ) AS t (plan text) );      PERFORM dblink_disconnect('LINK1');   --------------------------   BEGIN 	INSERT INTO log_query 	( 		query_md5hash , 		database_id ,  		timepoint , 		duration , 		query , 		explained_plan , 		plan_md5hash ,  		explained_plan_wo_costs ,  		plan_hash_value ,  		ip ,  		port , 		pid 	)  	VALUES  	( 		log_md5hash , 		log_database_id ,  		log_timepoint ,  		log_duration ,  		log_query , 		log_explain_plan ,  		md5(log_explain_plan::text) , 		log_plan_wo_costs ,  		md5(log_plan_wo_costs::text), 		client_ip ,  		client_port ,  		current_pid 		 	); 	activity_string = 	'New query has logged '|| 						' database_id = '|| log_database_id || 						' query_md5hash='||log_md5hash|| 						' , timepoint = '||to_char(log_timepoint,'YYYYMMDD HH24:MI:SS'); 	PERFORM pg_log( log_database_id , 'log_query' , activity_string);    	EXCEPTION 	  WHEN unique_violation THEN 		activity_string = 	'EXCEPTION *** query already has logged '|| 							' database_id = '|| log_database_id || 							' query_md5hash='||log_md5hash|| 							' , timepoint = '||to_char(log_timepoint,'YYYYMMDD HH24:MI:SS');					          PERFORM pg_log( log_database_id , 'log_query' , activity_string); 	END;  	SELECT 	queryid 	INTO   	log_queryid 	FROM 	log_query  	WHERE 	query_md5hash = log_md5hash AND 			timepoint = log_timepoint;  	IF log_queryid IS NOT NULL  	THEN  	  RETURN result; 	END IF; 	 	------------------------------------------------ 	SELECT *  	INTO log_query_rec 	FROM log_query 	WHERE query_md5hash = log_md5hash AND timepoint = log_timepoint ;  	 	log_query_rec.query=regexp_replace(log_query_rec.query,';+','','g');  	FOR pg_stat_history_rec IN 	 SELECT              queryid ,   	    query  	 FROM         pg_stat_db_queries       WHERE   	   database_id = log_database_id AND        queryid is not null  	LOOP 	  pg_stat_query = pg_stat_history_rec.query ;  	  pg_stat_query=regexp_replace(pg_stat_query,'\n+',' ','g'); 	  pg_stat_query=regexp_replace(pg_stat_query,'\t+',' ','g'); 	  pg_stat_query=regexp_replace(pg_stat_query,' +',' ','g'); 	  pg_stat_query=regexp_replace(pg_stat_query,'\$.','%','g'); 	 	  log_query_text = trim(trailing ' ' from log_query_rec.query); 	  pg_stat_query_text = pg_stat_query;    	   	  IF (log_query_text LIKE pg_stat_query_text) THEN 		found_flag = TRUE ; 	  ELSE 		found_flag = FALSE ; 	  END IF;	    	  IF found_flag THEN 	     		UPDATE log_query SET queryid = pg_stat_history_rec.queryid WHERE query_md5hash = log_md5hash AND timepoint = log_timepoint ; 		activity_string = 	' updated queryid = '||pg_stat_history_rec.queryid|| 		                    ' for log_query with id = '||log_query_rec.id;					 		EXIT ; 	  END IF ;	   	END LOOP ;	   RETURN result ; END $$ LANGUAGE plpgsql;

--update_history_pg_stat_activity_by_queryid.sql CREATE OR REPLACE FUNCTION update_history_pg_stat_activity_by_queryid() RETURNS boolean AS $$ DECLARE   result boolean ;   history_pg_stat_activity_rec record ;    pg_stat_query text ;   pg_stat_query_text text ;   pg_stat_history_rec record;   found_flag boolean;   history_pg_stat_activity_query text ;    query_text text ;   activity_string text ;     BEGIN   RAISE NOTICE '***update_history_pg_stat_activity_by_queryid';      result = TRUE ;      FOR history_pg_stat_activity_rec IN    SELECT DISTINCT(query) AS query   FROM activity_hist.history_pg_stat_activity   WHERE queryid IS NULL   LOOP 		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_rec.query,'\n+',' ','g'); 		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query,'\t+',' ','g'); 		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query,' +',' ','g'); 		history_pg_stat_activity_query = regexp_replace(history_pg_stat_activity_query,';','','g'); 		query_text = trim(trailing ' ' from history_pg_stat_activity_query); 		 		FOR pg_stat_history_rec IN 		SELECT  			queryid , 			query  		FROM  			--pg_stat_history 			pg_stat_db_queries 		WHERE   			queryid is not null  		GROUP BY queryid ,	query  		LOOP 			pg_stat_query = pg_stat_history_rec.query ;  			pg_stat_query=regexp_replace(pg_stat_query,'\n+',' ','g'); 			pg_stat_query=regexp_replace(pg_stat_query,'\t+',' ','g'); 			pg_stat_query=regexp_replace(pg_stat_query,' +',' ','g'); 			pg_stat_query=regexp_replace(pg_stat_query,'\$.','%','g');	 			 			pg_stat_query_text = pg_stat_query;  	   			IF (query_text LIKE pg_stat_query_text) THEN 				found_flag = TRUE ; 			ELSE 				found_flag = FALSE ; 			END IF;	   			 			IF found_flag  			THEN 				UPDATE activity_hist.history_pg_stat_activity 				SET queryid = pg_stat_history_rec.queryid 				WHERE regexp_replace(regexp_replace(regexp_replace(regexp_replace(query,'\n+',' ','g'),'\t+',' ','g'),' +',' ','g'),';','','g')  				      LIKE query_text||'%' ; 		 				activity_string = 	'history_pg_stat_activity has updated by queryid = '||pg_stat_history_rec.queryid; 				RAISE NOTICE '%',activity_string;	 				 				PERFORM pg_log( 999 , 'update_history_pg_stat_activity_by_queryid' , activity_string);  				 				EXIT ; 				 			END IF ;	   		END LOOP ; 		 		IF NOT found_flag  		THEN 			activity_string = 'WARNING : Not FOUND queryid for the query : '||query_text ; 			 			RAISE NOTICE '%',activity_string;	 				 			PERFORM pg_log( 999 , 'update_history_pg_stat_activity_by_queryid' , activity_string);  		END IF ; 		 	RAISE NOTICE 'UPDATE log_query if query has not logged in log-file'; 		   END LOOP;      RETURN result ; END $$ LANGUAGE plpgsql;