Issue - (#62): Logging CMD Type like SELECT, UPDATE, INSERT, DELETE, UNKNOWN.

PG-150

README.md

@@ -489,4 +489,24 @@ postgres=# select tables_names, query from pg_stat_monitor;
 (8 rows)
 ```
 
+### Query command Type (SELECT, UPDATE, DELETE, UPDATE, TRUNCATE, NONE)
 
+**`cmd_type`**: List the command type of the query.
+
+```
+postgres=# select substr(query,0, 50) as query, cmd_type from pg_stat_monitor where elevel = 0;
+                       query                       |                     cmd_type                     
+---------------------------------------------------+--------------------------------------------------
+ select substr(query,$1, $2) as query, cmd_type fr | {0," 0"," 0"," 1"," 0"," 0"," 0"," 0"," 0"," 1"}
+ vacuum pgbench_branches                           | {0," 0"," 0"," 1"," 0"," 0"," 0"," 0"," 0"," 0"}
+ vacuum analyze pgbench_branches                   | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ alter table pgbench_branches add primary key (bid | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ alter table pgbench_tellers add primary key (tid) | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ insert into pgbench_tellers(tid,bid,tbalance) val | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ UPDATE pgbench_accounts SET abalance = abalance + | {0," 1"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ insert into pgbench_tellers(tid,bid,tbalance) val | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ insert into pgbench_branches(bid,bbalance) values | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ truncate table pgbench_accounts, pgbench_branches | {0," 0"," 0"," 1"," 0"," 0"," 0"," 0"," 0"," 0"}
+ vacuum analyze pgbench_accounts                   | {1," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"," 0"}
+ create table pgbench_history(tid int,bid int,aid  | {0," 0"," 0"," 1"," 0"," 0"," 0"," 0"," 0"," 0"}
+```

pg_stat_monitor.c

@@ -184,7 +184,7 @@ _PG_init(void)
 	planner_hook            		= pgss_planner_hook;
 	emit_log_hook                   = pgsm_emit_log_hook;
 	prev_ExecutorCheckPerms_hook 	= ExecutorCheckPerms_hook;
-	ExecutorCheckPerms_hook		 	= pgss_ExecutorCheckPerms;
+	ExecutorCheckPerms_hook		= pgss_ExecutorCheckPerms;
 
 	system_init = true;
 }
@@ -461,8 +461,12 @@ pgss_ExecutorCheckPerms(List *rt, bool abort)
 {
 	ListCell *lr;
 	pgssSharedState *pgss = pgsm_get_ss();
+	int i;
+
+	LWLockAcquire(pgss->lock, LW_EXCLUSIVE);
+	for (i = 0; i < CMD_LST; i++)
+		pgss->cmdTag[i] = 0;
 
-	memset(pgss->cmdTag, 0x0, sizeof(int32) * 5);
 	foreach(lr, rt)
     {
         RangeTblEntry *rte = lfirst(lr);
@@ -473,7 +477,14 @@ pgss_ExecutorCheckPerms(List *rt, bool abort)
         else if (rte->requiredPerms & ACL_UPDATE) pgss->cmdTag[1] = true;
         else if (rte->requiredPerms & ACL_DELETE) pgss->cmdTag[2] = true;
         else if (rte->requiredPerms & ACL_SELECT) pgss->cmdTag[3] = true;
+        else if (rte->requiredPerms & ACL_TRUNCATE) pgss->cmdTag[4] = true;
+        else if (rte->requiredPerms & ACL_REFERENCES) pgss->cmdTag[5] = true;
+        else if (rte->requiredPerms & ACL_TRIGGER) pgss->cmdTag[6] = true;
+        else if (rte->requiredPerms & ACL_EXECUTE) pgss->cmdTag[7] = true;
+        else if (rte->requiredPerms & ACL_CREATE) pgss->cmdTag[8] = true;
+        else pgss->cmdTag[9] = true;
 	}
+	LWLockRelease(pgss->lock);
 
     if (prev_ExecutorCheckPerms_hook)
         return prev_ExecutorCheckPerms_hook(rt, abort);
@@ -910,13 +921,14 @@ static void pgss_store(uint64 queryId,
 			if (total_time > PGSM_RESPOSE_TIME_LOWER_BOUND + (PGSM_RESPOSE_TIME_STEP * MAX_RESPONSE_BUCKET))
 				e->counters.resp_calls[MAX_RESPONSE_BUCKET - 1]++;
 		}
-		for (i = 0; i < 5; i++)
+		for (i = 0; i < CMD_LST; i++)
 			e->counters.info.cmd_type[i] = pgss->cmdTag[i];
 
 		e->counters.error.elevel = elevel;
 		e->counters.error.sqlcode = sqlcode;
 		for(i = 0; i < message_len; i++)
 			e->counters.error.message[i] = message[i];
+
 		e->counters.calls[kind].rows += rows;
 		e->counters.blocks.shared_blks_hit += bufusage->shared_blks_hit;
 		e->counters.blocks.shared_blks_read += bufusage->shared_blks_read;
@@ -1100,7 +1112,7 @@ pg_stat_monitor_internal(FunctionCallInfo fcinfo,
 			else
 				nulls[i++] = true;
 		}
-		values[i++] = ArrayGetTextDatum(tmp.info.cmd_type, 5);
+		values[i++] = ArrayGetTextDatum(tmp.info.cmd_type, CMD_LST);
 		values[i++] = Int64GetDatumFast(tmp.error.elevel);
 		values[i++] = Int64GetDatumFast(tmp.error.sqlcode);
 		if (strlen(tmp.error.message) == 0)
@@ -2093,11 +2105,12 @@ static Datum
 array_get_datum(int32 arr[], int len)
 {
 	int     j;
-	char    str[1024] = {0};
+	char    str[1024];
 	char    tmp[10];
 	bool    first = true;
 
-	memset(str, 0, 1024);
+	memset(str, 0, 1023);
+
 	/* Need to calculate the actual size, and avoid unnessary memory usage */
 	for (j = 0; j < len; j++)
 	{
@@ -2112,6 +2125,7 @@ array_get_datum(int32 arr[], int len)
 		strcat(str,tmp);
 	}
 	return CStringGetTextDatum(str);
+
 }
 
 static uint64

pg_stat_monitor.h

@@ -61,6 +61,7 @@
 #define MAX_OBJECT_CACHE	100
 #define TEXT_LEN			255
 #define ERROR_MESSAGE_LEN	100
+#define CMD_LST				10
 
 typedef struct GucVariables
 {
@@ -130,7 +131,7 @@ typedef struct QueryInfo
 	Oid			dbid;						/* database OID */
 	uint		host;						/* client IP */
 	int64       type; 						/* type of query, options are query, info, warning, error, fatal */
-	int32		cmd_type[5];				/* query command type SELECT/UPDATE/DELETE/INSERT */
+	int32		cmd_type[CMD_LST];				/* query command type SELECT/UPDATE/DELETE/INSERT */
 	char		tables_name[MAX_REL_LEN];   /* table names involved in the query */
 } QueryInfo;
 
@@ -223,7 +224,7 @@ typedef struct pgssSharedState
 	uint64			bucket_overflow[MAX_BUCKETS];
 	uint64			bucket_entry[MAX_BUCKETS];
 	int				query_buf_size_bucket;
-	int				cmdTag[5];
+	int32			cmdTag[CMD_LST];
 	Timestamp		bucket_start_time[MAX_BUCKETS];   	/* start time of the bucket */
 } pgssSharedState;