Add logic to support ranged GTID updates from proxysql_mysqlbinlog.

Author: proton-lisandro-pin

include/proxysql_gtid.h

@@ -2,13 +2,39 @@
 #define PROXYSQL_GTID
 // highly inspired by libslave
 // https://github.com/vozbu/libslave/
+#include <list>
 #include <string>
 #include <unordered_map>
-#include <list>
 #include <utility>
 
 typedef std::pair<std::string, int64_t> gtid_t;
-typedef std::pair<int64_t, int64_t> gtid_interval_t;
+
+class Gtid_Interval {
+	public:
+		int64_t start;
+		int64_t end;
+
+	private:
+		void init(const int64_t _start, const int64_t _end);
+		void init(const char* s);
+
+	public:
+		explicit Gtid_Interval(const int64_t _start, const int64_t _end);
+		explicit Gtid_Interval(const char* s);
+		explicit Gtid_Interval(const std::string& s);
+
+		const std::string to_string(void);
+		const bool contains(const Gtid_Interval &iv);
+		const bool contains(int64_t gtid);
+		const bool merge(const Gtid_Interval& other);
+
+		const int cmp(const Gtid_Interval& other);
+		const bool operator<(const Gtid_Interval& other);
+		const bool operator==(const Gtid_Interval& other);
+};
+typedef Gtid_Interval gtid_interval_t;
+
+// TODO: make me a proper class.
 typedef std::unordered_map<std::string, std::list<gtid_interval_t>> gtid_set_t;
 
 /*
@@ -31,4 +57,4 @@ class Gtid_Server_Info {
 };
 */
 
-#endif /* PROXYSQL_GTID */
+#endif /* PROXYSQL_GTID */

lib/GTID_Server_Data.cpp

@@ -269,7 +269,7 @@ bool GTID_Server_Data::gtid_exists(char *gtid_uuid, uint64_t gtid_trxid) {
 		return false;
 	}
 	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr) {
-		if ((int64_t)gtid_trxid >= itr->first && (int64_t)gtid_trxid <= itr->second) {
+		if (itr->contains((int64_t)gtid_trxid)) {
 //			fprintf(stderr,"YES\n");
 			return true;
 		}
@@ -412,12 +412,7 @@ std::string gtid_executed_to_string(gtid_set_t& gtid_executed) {
 		s.insert(23,"-");
 		s = s + ":";
 		for (auto itr = it->second.begin(); itr != it->second.end(); ++itr) {
-			std::string s2 = s;
-			s2 = s2 + std::to_string(itr->first);
-			s2 = s2 + "-";
-			s2 = s2 + std::to_string(itr->second);
-			s2 = s2 + ",";
-			gtid_set = gtid_set + s2;
+			gtid_set += s + itr->to_string() + ",";
 		}
 	}
 	// Extract latest comma only in case 'gtid_executed' isn't empty
@@ -428,52 +423,52 @@ std::string gtid_executed_to_string(gtid_set_t& gtid_executed) {
 }
 
 
-
-void addGtid(const gtid_t& gtid, gtid_set_t& gtid_executed) {
-	auto it = gtid_executed.find(gtid.first);
-	if (it == gtid_executed.end())
-	{
-		gtid_executed[gtid.first].emplace_back(gtid.second, gtid.second);
-		return;
+// Merges a GTID interval into a GTID set instance.
+// Returns true if the GTID set was modified, false otherwise.
+bool addGtid(const std::string& uuid, const gtid_interval_t &iv, gtid_set_t& gtid_executed) {
+	auto it = gtid_executed.find(uuid);
+	if (it == gtid_executed.end()) {
+		// new UUID entry
+		gtid_executed[uuid].emplace_back(iv);
+		return true;
 	}
 
-	bool flag = true;
-	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr)
-	{
-		if (gtid.second >= itr->first && gtid.second <= itr->second)
-			return;
-		if (gtid.second + 1 == itr->first)
-		{
-			--itr->first;
-			flag = false;
-			break;
+	// insert/merge GTID interval
+	auto pos = it->second.begin();
+	for (; pos != it->second.end(); ++pos) {
+		if (pos->contains(iv)) {
+			// nothing to do
+			return false;
 		}
-		else if (gtid.second == itr->second + 1)
-		{
-			++itr->second;
-			flag = false;
+		if (pos->merge(iv))
 			break;
-		}
-		else if (gtid.second < itr->first)
-		{
-			it->second.emplace(itr, gtid.second, gtid.second);
-			return;
-		}
+	}
+	if (pos == it->second.end()) {
+		it->second.emplace_back(iv);
 	}
 
-	if (flag)
-		it->second.emplace_back(gtid.second, gtid.second);
-
-	for (auto itr = it->second.begin(); itr != it->second.end(); ++itr)
-	{
-		auto next_itr = std::next(itr);
-		if (next_itr != it->second.end() && itr->second + 1 == next_itr->first)
-		{
-			itr->second = next_itr->second;
-			it->second.erase(next_itr);
+	// merge overlapping GTID ranges, if any
+	it->second.sort();
+	auto a = it->second.begin();
+	while (a != it->second.end()) {
+		auto b = std::next(a);
+		if (b == it->second.end()) {
 			break;
 		}
+		if (a->merge(*b)) {
+				it->second.erase(b);
+				continue;
+		}
+		a++;
 	}
+
+	return true;
+}
+
+// Merges a single GTID into a gitd_executed instance.
+inline bool addGtid(const gtid_t& gtid, gtid_set_t& gtid_executed) {
+	gtid_interval_t iv = Gtid_Interval(gtid.second, gtid.second);
+	return addGtid(gtid.first, iv, gtid_executed);
 }
 
 /**
@@ -499,41 +494,9 @@ void addGtid(const gtid_t& gtid, gtid_set_t& gtid_executed) {
  * @note This function is critical for maintaining accurate GTID metrics across
  *       binlog reader reconnections and preventing events_count resets.
  */
-bool addGtidInterval(gtid_set_t& gtid_executed, std::string server_uuid, int64_t txid_start, int64_t txid_end) {
-	bool updated = true;
-
-	auto it = gtid_executed.find(server_uuid);
-	if (it == gtid_executed.end()) {
-		gtid_executed[server_uuid].emplace_back(txid_start, txid_end);
-		return updated;
-	}
-
-	bool insert = true;
-
-	// When ProxySQL reconnects with binlog reader, it might
-	// receive updated txid intervals in the bootstrap message.
-	// For example,
-	// before disconnection -> server_UUID:1-10
-	// after reconnection   -> server_UUID:1-19
-	auto &txid_intervals = it->second;
-	for (auto &interval : txid_intervals) {
-		if (interval.first == txid_start) {
-			if(interval.second == txid_end) {
-				updated = false;
-			} else {
-				interval.second = txid_end;
-			}
-			insert = false;
-			break;
-		}
-	}
-
-	if (insert) {
-		txid_intervals.emplace_back(txid_start, txid_end);
-
-	}
-
-	return updated;
+inline bool addGtidInterval(gtid_set_t& gtid_executed, std::string server_uuid, int64_t txid_start, int64_t txid_end) {
+	gtid_interval_t iv = Gtid_Interval(txid_start, txid_end);
+	return addGtid(server_uuid, iv, gtid_executed);
 }
 
 void * GTID_syncer_run() {

lib/Makefile

@@ -71,6 +71,7 @@ _OBJ_CXX := ProxySQL_GloVars.oo network.oo debug.oo configfile.oo Query_Cache.oo
 	ProxySQL_Admin_Tests.oo ProxySQL_Admin_Tests2.oo ProxySQL_Admin_Scheduler.oo ProxySQL_Admin_Disk_Upgrade.oo ProxySQL_Admin_Stats.oo \
 	Admin_Handler.oo Admin_FlushVariables.oo Admin_Bootstrap.oo \
 	Base_Session.oo Base_Thread.oo \
+	proxysql_gtid.oo \
 	proxy_protocol_info.oo \
 	proxysql_find_charset.oo ProxySQL_Poll.oo \
 	PgSQL_Protocol.oo PgSQL_Thread.oo PgSQL_Data_Stream.oo PgSQL_Session.oo PgSQL_Variables.oo PgSQL_HostGroups_Manager.oo PgSQL_Connection.oo PgSQL_Backend.oo PgSQL_Logger.oo PgSQL_Authentication.oo PgSQL_Error_Helper.oo \

lib/proxysql_gtid.cpp

@@ -0,0 +1,116 @@
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <string>
+
+#include "proxysql_gtid.h"
+
+
+// Initializes a GTID interval.
+// Implemented via private method as C++03 does not support delegated constructors :(
+void Gtid_Interval::init(const int64_t _start, const int64_t _end) {
+	start = _start;
+	end = _end;
+
+	if (start > end) {
+		std::swap(start, end);
+	}
+}
+
+// Initializes a GTID interval from a string buffer, in [gtid]{-[gtid]} format.
+void Gtid_Interval::init(const char *s) {
+	uint64_t _start, _end;
+
+	if (sscanf(s, "%lu-%lu", &_start, &_end) == 2) {
+		init((int64_t)_start, (int64_t)_end);
+		return;
+	}
+	if (sscanf(s, "%lu", &_start) == 1) {
+		init((int64_t)_start, (int64_t)_start);
+		return;
+	}
+
+	init(0, 0);
+}
+
+
+Gtid_Interval::Gtid_Interval(const int64_t _start, const int64_t _end) {
+	init(_start, _end);
+}
+
+Gtid_Interval::Gtid_Interval(const char *s) {
+	init(s);
+}
+
+Gtid_Interval::Gtid_Interval(const std::string& s) {
+	init(s.c_str());
+}
+
+// Checks if a given GTID interval is contained in this interval.
+const bool Gtid_Interval::contains(const Gtid_Interval &iv) {
+	return (iv.start >= start && iv.end <= end);
+}
+
+// Checks if a given GTID is contained in this interval.
+const bool Gtid_Interval::contains(int64_t gtid) {
+	return (gtid >= start && gtid <= end);
+}
+
+// Yields a string representation for a GTID interval.
+const std::string Gtid_Interval::to_string(void) {
+	if (start == end) {
+		return std::to_string(start);
+	}
+	return std::to_string(start) + "-" + std::to_string(end);
+}
+
+// Attempts to merge two GTID intervals. Returns true if the intervals were merged (and potentially modified), false otherwise.
+const bool Gtid_Interval::merge(const Gtid_Interval& other) {
+	if (other.start >= start && other.end <= end) {
+		// other is contained by interval
+		return true;
+	}
+	if (other.start <= start && other.end >= end) {
+		// other contains whole of existing interval
+		start = other.start;
+		end = other.end;
+		return true;
+	}
+	if (other.start <= start && other.end >= (start-1)) {
+		// other overlaps interval at start
+		start = other.start;
+		return true;
+	}
+	if (other.end >= end && other.start <= (end+1)) {
+		// other overlaps interval at end
+		end = other.end;
+		return true;
+	}
+
+	return false;
+}
+
+// Comapres two GTID intervals, by strict weak ordering.
+const int Gtid_Interval::cmp(const Gtid_Interval& other) {
+	if (start < other.start) {
+		return -1;
+	}
+	if (start > other.start) {
+		return 1;
+	}
+	if (end < other.end) {
+		return -1;
+	}
+	if (end > other.end) {
+		return 1;
+	}
+	return 0;
+}
+
+const bool Gtid_Interval::operator<(const Gtid_Interval& other) {
+	return cmp(other) == -1;
+}
+
+const bool Gtid_Interval::operator==(const Gtid_Interval& other) {
+	return cmp(other) == 0;
+}