feat(stats): Purge query digest based on last seen

wazir-ahmed · wazir-ahmed · commit 4fa9a4740766 · 2025-05-01T13:23:19.000+05:30
- Add handler for purge query digest command
- Handle last_seen argument in purge query digest operations
- Make purge logic common for both sync &amp; async operations

Signed-off-by: Wazir Ahmed &lt;wazirahmedf@gmail.com&gt;
diff --git a/include/gen_utils.h b/include/gen_utils.h
@@ -339,6 +339,8 @@ bool mywildcmp(const char *p, const char *str);
 std::string trim(const std::string& s);
 char* escape_string_single_quotes_and_backslashes(char* input, bool free_it);
 const char* escape_string_backslash_spaces(const char* input);
+time_t monotonic_time_to_realtime(time_t mt);
+time_t realtime_to_monotonic_time(time_t rt);
 
 /**
  * @brief Helper function that converts a MYSQL_RES into a 'SQLite3_result'.
diff --git a/include/query_processor.h b/include/query_processor.h
@@ -282,7 +282,7 @@ class Query_Processor {
 	std::pair<SQLite3_result*,int> get_query_digests_v2(const bool use_resultset = true);
 	std::pair<SQLite3_result*,int> get_query_digests_reset_v2(const bool copy, const bool use_resultset = true);
 	void get_query_digests_reset(umap_query_digest* uqd, umap_query_digest_text* uqdt);
-	unsigned long long purge_query_digests(bool async_purge, bool parallel, char** msg);
+	unsigned long long purge_query_digests(bool async_purge, bool parallel, time_t last_seen = 0);
 
 	void save_query_rules(SQLite3_result* resultset);
 
@@ -394,8 +394,8 @@ class Query_Processor {
 	DEFINE_HAS_METHOD_STRUCT(query_parser_first_comment_extended);
 	DEFINE_HAS_METHOD_STRUCT(process_query_extended);
 
-	unsigned long long purge_query_digests_async(char** msg);
-	unsigned long long purge_query_digests_sync(bool parallel);
+	unsigned long long purge_query_digests_async(time_t last_seen = 0);
+	unsigned long long purge_query_digests_sync(bool parallel, time_t last_seen = 0);
 
 	/**
 	 * @brief Searches for a matching rule in the supplied map, returning the destination hostgroup.
diff --git a/lib/Admin_Bootstrap.cpp b/lib/Admin_Bootstrap.cpp
@@ -112,9 +112,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;
diff --git a/lib/Admin_FlushVariables.cpp b/lib/Admin_FlushVariables.cpp
@@ -112,9 +112,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;
diff --git a/lib/Admin_Handler.cpp b/lib/Admin_Handler.cpp
@@ -114,9 +114,6 @@ extern struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;
@@ -276,6 +273,17 @@ const std::vector<std::string> LOAD_COREDUMP_FROM_MEMORY = {
 	"LOAD COREDUMP TO RUNTIME" ,
 	"LOAD COREDUMP TO RUN" };
 
+const std::vector<std::string> CMD_PREFIX_PURGE_QUERY_DIGESTS = {
+	"PURGE TABLE stats.stats_mysql_query_digest TO ",
+	"PURGE TABLE stats_mysql_query_digest TO ",
+	"PURGE stats.stats_mysql_query_digest TO ",
+	"PURGE stats_mysql_query_digest TO ",
+	"PURGE TABLE stats.stats_pgsql_query_digest TO ",
+	"PURGE TABLE stats_pgsql_query_digest TO ",
+	"PURGE stats.stats_pgsql_query_digest TO ",
+	"PURGE stats_pgsql_query_digest TO ",
+};
+
 extern unordered_map<string,std::tuple<string, vector<string>, vector<string>>> load_save_disk_commands;
 
 bool is_admin_command_or_alias(const std::vector<std::string>& cmds, char *query_no_space, int query_no_space_length) {
@@ -288,6 +296,17 @@ bool is_admin_command_or_alias(const std::vector<std::string>& cmds, char *query
 	return false;
 }
 
+const char * match_command_prefix(const std::vector<std::string>& cmd_prefix, char *query, int query_len) {
+	for (auto &prefix : cmd_prefix) {
+		if ((unsigned int) query_len >= prefix.length()
+			&& !strncasecmp(prefix.c_str(), query, prefix.length()))
+		{
+			return prefix.c_str();
+		}
+	}
+
+	return nullptr;
+}
 
 template <typename S>
 bool FlushCommandWrapper(S* sess, const std::vector<std::string>& cmds, char *query_no_space, int query_no_space_length, const string& name, const string& direction) {
@@ -322,6 +341,35 @@ bool FlushCommandWrapper(S* sess, const string& modname, char *query_no_space, i
 	return false;
 }
 
+std::tuple<bool, enum SERVER_TYPE, time_t> parse_command_purge_query_digests(char *query, int query_len) {
+	bool match = false;
+	enum SERVER_TYPE server_type = SERVER_TYPE_MYSQL;
+	time_t last_seen = 0;
+
+	const char *prefix = match_command_prefix(CMD_PREFIX_PURGE_QUERY_DIGESTS, query, query_len);
+	if (prefix) {
+		match = true;
+
+		if (strstr(prefix, "_pgsql_") != nullptr) {
+			server_type = SERVER_TYPE_PGSQL;
+		}
+
+		// parse timestamp
+		char *ts_str = strdup(query + strlen(prefix));
+		char *ts_end = nullptr;
+		long long ts = strtoll(trim_spaces_in_place(ts_str), &ts_end, 10);
+
+		// ts_str should only contain digits and respresent a valid timestamp
+		if ((*ts_end == 0) && (ts > 0)) {
+			last_seen = realtime_to_monotonic_time(ts);
+		}
+
+		free(ts_str);
+	}
+
+	return std::make_tuple(match, server_type, last_seen);
+}
+
 template <typename S>
 bool admin_handler_command_kill_connection(char *query_no_space, unsigned int query_no_space_length, S* sess, ProxySQL_Admin *pa) {
 	uint32_t id=atoi(query_no_space+16);
@@ -2498,10 +2546,8 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 					SPA->admindb->execute("DELETE FROM stats.stats_mysql_query_digest_reset");
 					SPA->vacuum_stats(true);
 					// purge the digest map, asynchronously, in single thread
-					char *msg = NULL;
-					int r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, &msg);
-					SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-					free(msg);
+					int r1 = GloMyQPro->purge_query_digests(true, false);
+					SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 					run_query=false;
 					goto __run_query;
 				}
@@ -2534,16 +2580,45 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 					SPA->admindb->execute("DELETE FROM stats.stats_pgsql_query_digest_reset");
 					SPA->vacuum_stats(true);
 					// purge the digest map, asynchronously, in single thread
-					char* msg = NULL;
-					int r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(true, false, &msg);
-					SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-					free(msg);
+					int r1 = GloPgQPro->purge_query_digests(true, false);
+					SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 					run_query = false;
 					goto __run_query;
 				}
 			}
 		}
 	}
+
+	// handles 'PURGE stats_mysql_query_digest TO <value>'.
+	// any entry in stats_mysql_query_digest where last_seen is less than <value> will be deleted.
+	if (!strncasecmp("PURGE ", query_no_space, strlen("PURGE "))
+		&& sess->session_type == PROXYSQL_SESSION_ADMIN
+	) {
+		auto result = parse_command_purge_query_digests(query_no_space, query_no_space_length);
+		bool match = std::get<0>(result);
+
+		if (match == true) {
+			int ret = 0;
+			enum SERVER_TYPE type = std::get<1>(result);
+			time_t last_seen = std::get<2>(result);
+
+			if (last_seen > 0) {
+				if (type == SERVER_TYPE_MYSQL) {
+					ret = GloMyQPro->purge_query_digests(true, false, last_seen);
+				} else if (type == SERVER_TYPE_PGSQL) {
+					ret = GloPgQPro->purge_query_digests(true, false, last_seen);
+				}
+
+				pa->send_ok_msg_to_client(sess, NULL, ret, query_no_space);
+			} else {
+				pa->send_error_msg_to_client(sess, "Invalid timestamp");
+			}
+
+			run_query = false;
+			goto __run_query;
+		}
+	}
+
 #ifdef DEBUG
 	/**
 	 * @brief Handles the 'PROXYSQL_SIMULATOR' command. Performing the operation specified in the payload
@@ -3837,4 +3912,3 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 // Explicitly instantiate the required template class and member functions
 template void admin_session_handler<MySQL_Session>(MySQL_Session* sess, void *_pa, PtrSize_t *pkt);
 template void admin_session_handler<PgSQL_Session>(PgSQL_Session* sess, void *_pa, PtrSize_t *pkt);
-
diff --git a/lib/ProxySQL_Admin.cpp b/lib/ProxySQL_Admin.cpp
@@ -114,9 +114,6 @@ struct MHD_Daemon *Admin_HTTP_Server;
 
 extern ProxySQL_Statistics *GloProxyStats;
 
-template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
-
 extern char *ssl_key_fp;
 extern char *ssl_cert_fp;
 extern char *ssl_ca_fp;
@@ -8465,4 +8462,3 @@ void ProxySQL_Admin::enable_replicationlag_testing() {
 	mysql_servers_wrunlock();
 }
 #endif // TEST_REPLICATIONLAG
-
diff --git a/lib/ProxySQL_Admin_Tests.cpp b/lib/ProxySQL_Admin_Tests.cpp
@@ -1,3 +1,4 @@
+#include <ctime>
 #include <iostream>     // std::cout
 #include <sstream>      // std::stringstream
 #include <fstream>
@@ -56,12 +57,12 @@ bool ProxySQL_Test___Refresh_MySQL_Variables(unsigned int cnt) {
 }
 
 template <enum SERVER_TYPE ST>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg) {
+int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, time_t last_seen) {
 	int r = 0;
 	if constexpr (ST == SERVER_TYPE_MYSQL) {
-		r = GloMyQPro->purge_query_digests(async_purge, parallel, msg);
+		r = GloMyQPro->purge_query_digests(async_purge, parallel, last_seen);
 	} else if constexpr (ST == SERVER_TYPE_PGSQL) {
-		r = GloPgQPro->purge_query_digests(async_purge, parallel, msg);
+		r = GloPgQPro->purge_query_digests(async_purge, parallel, last_seen);
 	}
 	return r;
 }
@@ -147,5 +148,5 @@ int ProxySQL_Test___GenerateRandomQueryInDigestTable(int n) {
 	return n*1000;
 }
 
-template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(bool async_purge, bool parallel, char** msg);
-template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(bool async_purge, bool parallel, char** msg);
+template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(bool async_purge, bool parallel, time_t last_seen);
+template int ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_PGSQL>(bool async_purge, bool parallel, time_t last_seen);
diff --git a/lib/ProxySQL_Admin_Tests2.cpp b/lib/ProxySQL_Admin_Tests2.cpp
@@ -43,7 +43,7 @@ static void init_rand_del() {
 int ProxySQL_Test___GetDigestTable(bool reset, bool use_swap);
 bool ProxySQL_Test___Refresh_MySQL_Variables(unsigned int cnt);
 template<enum SERVER_TYPE>
-int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, char **msg);
+int ProxySQL_Test___PurgeDigestTable(bool async_purge, bool parallel, time_t last_seen);
 int ProxySQL_Test___GenerateRandomQueryInDigestTable(int n);
 
 void ProxySQL_Admin::map_test_mysql_firewall_whitelist_rules_cleanup() {
@@ -590,21 +590,20 @@ void ProxySQL_Admin::ProxySQL_Test_Handler(ProxySQL_Admin *SPA, S* sess, char *q
 				break;
 			case 4:
 				// purge the digest map, synchronously, in single thread
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, false, NULL);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, false, test_arg1);
 				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 5:
 				// purge the digest map, synchronously, in multiple threads
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, true, NULL);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(false, true, test_arg1);
 				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 6:
 				// purge the digest map, asynchronously, in single thread
-				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, &msg);
-				SPA->send_ok_msg_to_client(sess, msg, r1, query_no_space);
-				free(msg);
+				r1 = ProxySQL_Test___PurgeDigestTable<SERVER_TYPE_MYSQL>(true, false, test_arg1);
+				SPA->send_ok_msg_to_client(sess, NULL, r1, query_no_space);
 				run_query=false;
 				break;
 			case 7:
diff --git a/lib/QP_query_digest_stats.cpp b/lib/QP_query_digest_stats.cpp
@@ -1,3 +1,4 @@
+#include "gen_utils.h"
 #include "query_processor.h"
 
 // reverse:  reverse string s in place
@@ -151,16 +152,13 @@ char **QP_query_digest_stats::get_row(umap_query_digest_text *digest_text_umap,
 	my_itoa(qdsp->count_star, count_star);
 	pta[5]=qdsp->count_star;
 
-	time_t __now;
-	time(&__now);
-	unsigned long long curtime=monotonic_time();
 	time_t seen_time;
-	seen_time= __now - curtime/1000000 + first_seen/1000000;
+	seen_time=monotonic_time_to_realtime(first_seen);
 	//sprintf(qdsp->first_seen,"%ld", seen_time);
 	my_itoa(qdsp->first_seen, seen_time);
 	pta[6]=qdsp->first_seen;
 
-	seen_time= __now - curtime/1000000 + last_seen/1000000;
+	seen_time=monotonic_time_to_realtime(last_seen);
 	//sprintf(qdsp->last_seen,"%ld", seen_time);
 	my_itoa(qdsp->last_seen, seen_time);
 	pta[7]=qdsp->last_seen;
@@ -187,4 +185,3 @@ char **QP_query_digest_stats::get_row(umap_query_digest_text *digest_text_umap,
 	pta[13]=qdsp->rows_sent;
 	return pta;
 }
-
diff --git a/lib/Query_Processor.cpp b/lib/Query_Processor.cpp
diff --git a/lib/gen_utils.cpp b/lib/gen_utils.cpp
