PostgreSQL Cluster Sync Fixes and Refactoring

doc/proxysql_cluster/proxysql_cluster_working.md

@@ -1,6 +1,9 @@
 # Introduction
 This documentation provides an in-depth look at the internal workings of the ProxySQL Cluster feature. It is intended for readers who are already familiar with the basic concepts and functionality of ProxySQL Cluster.
 
+For PostgreSQL-specific branch status and implementation notes related to PR
+`#5297`, see `doc/proxysql_cluster/pgsql_cluster_sync_pr5297_status.md`.
+
 # Prerequisites
 Before reading this documentation, it is mandatory that the reader has gone through the official ProxySQL Cluster documentation available at [https://proxysql.com/documentation/proxysql-cluster/](https://proxysql.com/documentation/proxysql-cluster/). This will provide the necessary background knowledge and understanding of terminologies to understand the internal workings of the feature.
 
@@ -219,4 +222,4 @@ graph TD
     CLOSE_CONNECTION[Close Connection] --> END
     END[End]
 end
-```
+```

include/proxysql_admin.h

@@ -313,6 +313,10 @@ class ProxySQL_Admin {
 		int cluster_mysql_variables_diffs_before_sync;
 		int cluster_admin_variables_diffs_before_sync;
 		int cluster_ldap_variables_diffs_before_sync;
+		int cluster_pgsql_variables_diffs_before_sync;
+		int cluster_pgsql_query_rules_diffs_before_sync;
+		int cluster_pgsql_servers_diffs_before_sync;
+		int cluster_pgsql_users_diffs_before_sync;
 		int cluster_mysql_servers_sync_algorithm;
 		bool cluster_mysql_query_rules_save_to_disk;
 		bool cluster_mysql_servers_save_to_disk;
@@ -321,6 +325,10 @@ class ProxySQL_Admin {
 		bool cluster_mysql_variables_save_to_disk;
 		bool cluster_admin_variables_save_to_disk;
 		bool cluster_ldap_variables_save_to_disk;
+		bool cluster_pgsql_variables_save_to_disk;
+		bool cluster_pgsql_query_rules_save_to_disk;
+		bool cluster_pgsql_servers_save_to_disk;
+		bool cluster_pgsql_users_save_to_disk;
 		int stats_mysql_connection_pool;
 		int stats_mysql_connections;
 		int stats_mysql_query_cache;
@@ -528,6 +536,7 @@ class ProxySQL_Admin {
 		bool checksum_mysql_variables;
 		bool checksum_admin_variables;
 		bool checksum_ldap_variables;
+		bool checksum_pgsql_variables;
 	} checksum_variables;
 	template<enum SERVER_TYPE pt>
 	void public_add_active_users(enum cred_username_type usertype, char *user=NULL) {
@@ -639,6 +648,7 @@ class ProxySQL_Admin {
 //	void flush_admin_variables__from_disk_to_memory(); // commented in 2.3 because unused
 	void flush_admin_variables__from_memory_to_disk();
 	void flush_ldap_variables__from_memory_to_disk();
+	void flush_pgsql_variables__from_memory_to_disk();
 	void load_mysql_servers_to_runtime(const incoming_servers_t& incoming_servers = {}, const runtime_mysql_servers_checksum_t& peer_runtime_mysql_server = {},
 		const mysql_servers_v2_checksum_t& peer_mysql_server_v2 = {});
 	void save_mysql_servers_from_runtime();

include/proxysql_glovars.hpp

@@ -3,6 +3,7 @@
 
 #define CLUSTER_SYNC_INTERFACES_ADMIN "('admin-mysql_ifaces','admin-restapi_port','admin-telnet_admin_ifaces','admin-telnet_stats_ifaces','admin-web_port','admin-pgsql_ifaces')"
 #define CLUSTER_SYNC_INTERFACES_MYSQL "('mysql-interfaces')"
+#define CLUSTER_SYNC_INTERFACES_PGSQL "('pgsql-interfaces')"
 
 #include <memory>
 #include <string.h>

lib/Admin_FlushVariables.cpp

@@ -400,6 +400,10 @@ void ProxySQL_Admin::flush_GENERIC_variables__checksum__database_to_runtime(cons
 		if (GloVars.cluster_sync_interfaces == false) {
 			q += " AND variable_name NOT IN " + string(CLUSTER_SYNC_INTERFACES_ADMIN);
 		}
+	} else if (modname == "pgsql") {
+		if (GloVars.cluster_sync_interfaces == false) {
+			q += " AND variable_name NOT IN " + string(CLUSTER_SYNC_INTERFACES_PGSQL);
+		}
 	}
 	q += " ORDER BY variable_name";
 	admindb->execute_statement(q.c_str(), &error , &cols , &affected_rows , &resultset);
@@ -415,6 +419,8 @@ void ProxySQL_Admin::flush_GENERIC_variables__checksum__database_to_runtime(cons
 		checkvar = &GloVars.checksums_values.mysql_variables;
 	} else if (modname == "ldap") {
 		checkvar = &GloVars.checksums_values.ldap_variables;
+	} else if (modname == "pgsql") {
+		checkvar = &GloVars.checksums_values.pgsql_variables;
 	}
 	assert(checkvar != NULL);
 	checkvar->set_checksum(buf);
@@ -889,49 +895,14 @@ void ProxySQL_Admin::flush_pgsql_variables___database_to_runtime(SQLite3DB* db,
 		GloPTH->commit();
 		GloPTH->wrunlock();
 
-		/* Checksums are always generated - 'admin-checksum_*' deprecated
-		{
-			// NOTE: 'GloPTH->wrunlock()' should have been called before this point to avoid possible
-			// deadlocks. See issue #3847.
-			pthread_mutex_lock(&GloVars.checksum_mutex);
-			// generate checksum for cluster
-			flush_mysql_variables___runtime_to_database(admindb, false, false, false, true, true);
-			char* error = NULL;
-			int cols = 0;
-			int affected_rows = 0;
-			SQLite3_result* resultset = NULL;
-			std::string q;
-			q = "SELECT variable_name, variable_value FROM runtime_global_variables WHERE variable_name LIKE 'mysql-\%' AND variable_name NOT IN ('mysql-threads')";
-			if (GloVars.cluster_sync_interfaces == false) {
-				q += " AND variable_name NOT IN " + string(CLUSTER_SYNC_INTERFACES_MYSQL);
-			}
-			q += " ORDER BY variable_name";
-			admindb->execute_statement(q.c_str(), &error, &cols, &affected_rows, &resultset);
-			uint64_t hash1 = resultset->raw_checksum();
-			uint32_t d32[2];
-			char buf[20];
-			memcpy(&d32, &hash1, sizeof(hash1));
-			snprintf(buf, sizeof(buf), "0x%0X%0X", d32[0], d32[1]);
-			GloVars.checksums_values.mysql_variables.set_checksum(buf);
-			GloVars.checksums_values.mysql_variables.version++;
-			time_t t = time(NULL);
-			if (epoch != 0 && checksum != "" && GloVars.checksums_values.mysql_variables.checksum == checksum) {
-				GloVars.checksums_values.mysql_variables.epoch = epoch;
-			}
-			else {
-				GloVars.checksums_values.mysql_variables.epoch = t;
+			{
+				// NOTE: 'GloPTH->wrunlock()' should have been called before this point to avoid possible
+				// deadlocks. See issue #3847.
+				pthread_mutex_lock(&GloVars.checksum_mutex);
+				flush_pgsql_variables___runtime_to_database(admindb, false, false, false, true, true);
+				flush_GENERIC_variables__checksum__database_to_runtime("pgsql", checksum, epoch);
+				pthread_mutex_unlock(&GloVars.checksum_mutex);
 			}
-			GloVars.epoch_version = t;
-			GloVars.generate_global_checksum();
-			GloVars.checksums_values.updates_cnt++;
-			pthread_mutex_unlock(&GloVars.checksum_mutex);
-			delete resultset;
-		}
-		proxy_info(
-			"Computed checksum for 'LOAD MYSQL VARIABLES TO RUNTIME' was '%s', with epoch '%llu'\n",
-			GloVars.checksums_values.mysql_variables.checksum, GloVars.checksums_values.mysql_variables.epoch
-		);
-		*/
 
 		/**
 		 * @brief Check and warn if TCP keepalive is disabled for PostgreSQL connections.

lib/Admin_Handler.cpp

@@ -3340,6 +3340,53 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 		}
 	}
 
+	if (sess->session_type == PROXYSQL_SESSION_ADMIN) { // no stats
+		string tn = "";
+		if (!strncasecmp(CLUSTER_QUERY_RUNTIME_PGSQL_SERVERS, query_no_space, strlen(CLUSTER_QUERY_RUNTIME_PGSQL_SERVERS))) {
+			tn = "cluster_pgsql_servers";
+		} else if (!strncasecmp(CLUSTER_QUERY_PGSQL_REPLICATION_HOSTGROUPS, query_no_space, strlen(CLUSTER_QUERY_PGSQL_REPLICATION_HOSTGROUPS))) {
+			tn = "pgsql_replication_hostgroups";
+		} else if (!strncasecmp(CLUSTER_QUERY_PGSQL_HOSTGROUP_ATTRIBUTES, query_no_space, strlen(CLUSTER_QUERY_PGSQL_HOSTGROUP_ATTRIBUTES))) {
+			tn = "pgsql_hostgroup_attributes";
+		} else if (!strncasecmp(CLUSTER_QUERY_PGSQL_SERVERS_V2, query_no_space, strlen(CLUSTER_QUERY_PGSQL_SERVERS_V2))) {
+			tn = "pgsql_servers_v2";
+		}
+		if (tn != "") {
+			GloAdmin->pgsql_servers_wrlock();
+			resultset = PgHGM->get_current_pgsql_table(tn);
+			GloAdmin->pgsql_servers_wrunlock();
+
+			if (resultset == nullptr) {
+				if (tn == "pgsql_servers_v2") {
+					const string query_empty_resultset {
+						string { PGHGM_GEN_CLUSTER_ADMIN_PGSQL_SERVERS } + " LIMIT 0"
+					};
+
+					char* error = NULL;
+					int cols = 0;
+					int affected_rows = 0;
+					proxy_debug(PROXY_DEBUG_MYSQL_CONNPOOL, 4, "%s\n", query);
+					GloAdmin->pgsql_servers_wrlock();
+					GloAdmin->admindb->execute_statement(query_empty_resultset.c_str(), &error, &cols, &affected_rows, &resultset);
+					GloAdmin->pgsql_servers_wrunlock();
+				} else {
+					resultset = PgHGM->dump_table_pgsql(tn);
+				}
+
+				if (resultset) {
+					sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+					delete resultset;
+					run_query = false;
+					goto __run_query;
+				}
+			} else {
+				sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+				run_query = false;
+				goto __run_query;
+			}
+		}
+	}
+
 	if (!strncasecmp(CLUSTER_QUERY_MYSQL_USERS, query_no_space, strlen(CLUSTER_QUERY_MYSQL_USERS))) {
 		if (sess->session_type == PROXYSQL_SESSION_ADMIN) {
 			pthread_mutex_lock(&users_mutex);
@@ -3353,6 +3400,19 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 		}
 	}
 
+	if (!strncasecmp(CLUSTER_QUERY_PGSQL_USERS, query_no_space, strlen(CLUSTER_QUERY_PGSQL_USERS))) {
+		if (sess->session_type == PROXYSQL_SESSION_ADMIN) {
+			pthread_mutex_lock(&users_mutex);
+			resultset = GloPgAuth->get_current_pgsql_users();
+			pthread_mutex_unlock(&users_mutex);
+			if (resultset != nullptr) {
+				sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+				run_query = false;
+				goto __run_query;
+			}
+		}
+	}
+
 	if (sess->session_type == PROXYSQL_SESSION_ADMIN) { // no stats
 		if (!strncasecmp(CLUSTER_QUERY_MYSQL_QUERY_RULES, query_no_space, strlen(CLUSTER_QUERY_MYSQL_QUERY_RULES))) {
 			GloMyQPro->wrlock();
@@ -3396,6 +3456,69 @@ void admin_session_handler(S* sess, void *_pa, PtrSize_t *pkt) {
 		}
 	}
 
+	if (sess->session_type == PROXYSQL_SESSION_ADMIN) { // no stats
+		if (!strncasecmp(CLUSTER_QUERY_PGSQL_QUERY_RULES, query_no_space, strlen(CLUSTER_QUERY_PGSQL_QUERY_RULES))) {
+			GloPgQPro->wrlock();
+			resultset = GloPgQPro->get_current_query_rules_inner();
+			if (resultset == NULL) {
+				GloPgQPro->wrunlock(); // unlock first
+				resultset = GloPgQPro->get_current_query_rules();
+				if (resultset) {
+					sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+					delete resultset;
+					run_query = false;
+					goto __run_query;
+				}
+			} else {
+				sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+				// DO NOT DELETE: this is the inner resultset of Query_Processor
+				GloPgQPro->wrunlock();
+				run_query = false;
+				goto __run_query;
+			}
+		}
+		if (!strncasecmp(CLUSTER_QUERY_PGSQL_QUERY_RULES_FAST_ROUTING, query_no_space, strlen(CLUSTER_QUERY_PGSQL_QUERY_RULES_FAST_ROUTING))) {
+			GloPgQPro->wrlock();
+			resultset = GloPgQPro->get_current_query_rules_fast_routing_inner();
+			if (resultset == NULL) {
+				GloPgQPro->wrunlock(); // unlock first
+				resultset = GloPgQPro->get_current_query_rules_fast_routing();
+				if (resultset) {
+					sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+					delete resultset;
+					run_query = false;
+					goto __run_query;
+				}
+			} else {
+				sess->SQLite3_to_MySQL(resultset, error, affected_rows, &sess->client_myds->myprot);
+				// DO NOT DELETE: this is the inner resultset of Query_Processor
+				GloPgQPro->wrunlock();
+				run_query = false;
+				goto __run_query;
+			}
+		}
+	}
+
+	if (!strncasecmp(CLUSTER_QUERY_PGSQL_VARIABLES, query_no_space, strlen(CLUSTER_QUERY_PGSQL_VARIABLES))) {
+		if (sess->session_type == PROXYSQL_SESSION_ADMIN) {
+			pthread_mutex_lock(&GloVars.checksum_mutex);
+			GloAdmin->flush_pgsql_variables___runtime_to_database(GloAdmin->admindb, false, false, false, true, true);
+			pthread_mutex_unlock(&GloVars.checksum_mutex);
+
+			l_free(query_length, query);
+			string q {
+				"SELECT variable_name, variable_value FROM runtime_global_variables WHERE variable_name LIKE 'pgsql-%'"
+			};
+			if (GloVars.cluster_sync_interfaces == false) {
+				q += " AND variable_name NOT IN " + string(CLUSTER_SYNC_INTERFACES_PGSQL);
+			}
+			q += " ORDER BY variable_name";
+			query = l_strdup(q.c_str());
+			query_length = strlen(query) + 1;
+			goto __run_query;
+		}
+	}
+
 	// if the client simply executes:
 	// SELECT COUNT(*) FROM runtime_mysql_query_rules_fast_routing
 	// we just return the count

lib/PgSQL_Variables_Validator.cpp

@@ -639,3 +639,33 @@ const pgsql_variable_validator pgsql_variable_validator_search_path = {
 	.validate = &pgsql_variable_validate_search_path,
 	.params = {}
 };
+
+/**
+ * @brief Validates an integer variable for PostgreSQL.
+ *
+ * This function checks if the provided value is a valid integer representation
+ * and falls within the specified range. The range is defined by the params
+ * parameter.
+ *
+ * @param value The value to validate.
+ * @param params The parameter structure containing the integer range.
+ * @param session Unused parameter.
+ * @param transformed_value If not null, will be set to null.
+ * @return true if the value is a valid integer representation within the specified range, false otherwise.
+ */
+bool pgsql_variable_validate_integer(const char* value, const params_t* params, PgSQL_Session* session, char** transformed_value) {
+   (void)session;
+   if (transformed_value) *transformed_value = nullptr;
+   char* end = nullptr;
+   errno = 0;
+   long num = strtol(value, &end, 10);
+   if (end == value || *end != '\0' || errno == ERANGE) return false;
+   if (num < params->int_range.min || num > params->int_range.max) return false;
+   return true;
+}
+
+const pgsql_variable_validator pgsql_variable_validator_integer = {
+	.type = VARIABLE_TYPE_INT,
+	.validate = &pgsql_variable_validate_integer,
+	.params = {}
+};