Added metadata to 'Query_Cache' entries for supporting EOF deprecation

include/MySQL_Protocol.h

@@ -59,6 +59,7 @@ class MySQL_Prepared_Stmt_info {
 	MySQL_Prepared_Stmt_info(unsigned char *, unsigned int);
 };
 
+uint8_t mysql_decode_length(unsigned char *ptr, uint64_t *len);
 
 class MySQL_Protocol {
 	private:

include/query_cache.hpp

@@ -30,6 +30,9 @@ struct __QC_entry_t {
 	unsigned long long create_ms; // when the entry was created, monotonic, millisecond granularity
 	unsigned long long expire_ms; // when the entry will expire, monotonic , millisecond granularity
 	unsigned long long access_ms; // when the entry was read last , monotonic , millisecond granularity
+	uint32_t column_eof_pkt_offset = 0;
+	uint32_t row_eof_pkt_offset = 0;
+	uint32_t ok_pkt_offset = 0;
 	uint32_t ref_count; // reference counter
 };
 
@@ -85,6 +88,7 @@ class Query_Cache {
 	Query_Cache();
 	~Query_Cache();
 	void print_version();
+	bool set(uint64_t user_hash, const unsigned char *kp, uint32_t kl, unsigned char *vp, uint32_t vl, unsigned long long create_ms, unsigned long long curtime_ms, unsigned long long expire_ms, bool deprecate_eof_active);
 	bool set(uint64_t , const unsigned char *, uint32_t, unsigned char *, uint32_t, unsigned long long, unsigned long long, unsigned long long);
 	unsigned char * get(uint64_t , const unsigned char *, const uint32_t, uint32_t *, unsigned long long, unsigned long long);
 	uint64_t flush();

lib/MySQL_Session.cpp

@@ -6078,6 +6078,7 @@ void MySQL_Session::MySQL_Result_to_MySQL_wire(MYSQL *mysql, MySQL_ResultSet *My
 						client_myds->resultset_length=MyRS->resultset_size;
 						unsigned char *aa=client_myds->resultset2buffer(false);
 						while (client_myds->resultset->len) client_myds->resultset->remove_index(client_myds->resultset->len-1,NULL);
+						bool deprecate_eof_active = client_myds->myconn->options.client_flag & CLIENT_DEPRECATE_EOF;
 						GloQC->set(
 							client_myds->myconn->userinfo->hash ,
 							(const unsigned char *)CurrentQuery.QueryPointer,
@@ -6086,7 +6087,8 @@ void MySQL_Session::MySQL_Result_to_MySQL_wire(MYSQL *mysql, MySQL_ResultSet *My
 							client_myds->resultset_length ,
 							thread->curtime/1000 ,
 							thread->curtime/1000 ,
-							thread->curtime/1000 + qpo->cache_ttl
+							thread->curtime/1000 + qpo->cache_ttl,
+							deprecate_eof_active
 						);
 						l_free(client_myds->resultset_length,aa);
 						client_myds->resultset_length=0;

lib/Query_Cache.cpp

@@ -6,6 +6,7 @@
 #include "proxysql_atomic.h"
 #include "SpookyV2.h"
 #include "prometheus_helpers.h"
+#include "MySQL_Protocol.h"
 
 #define THR_UPDATE_CNT(__a, __b, __c, __d) \
 	do {\
@@ -486,11 +487,53 @@ unsigned char * Query_Cache::get(uint64_t user_hash, const unsigned char *kp, co
 	return result;
 }
 
-bool Query_Cache::set(uint64_t user_hash, const unsigned char *kp, uint32_t kl, unsigned char *vp, uint32_t vl, unsigned long long create_ms, unsigned long long curtime_ms, unsigned long long expire_ms) {
+bool Query_Cache::set(uint64_t user_hash, const unsigned char *kp, uint32_t kl, unsigned char *vp, uint32_t vl, unsigned long long create_ms, unsigned long long curtime_ms, unsigned long long expire_ms, bool deprecate_eof_active) {
 	QC_entry_t *entry = (QC_entry_t *)malloc(sizeof(QC_entry_t));
 	entry->klen=kl;
 	entry->length=vl;
 	entry->ref_count=0;
+	entry->column_eof_pkt_offset=0;
+	entry->row_eof_pkt_offset=0;
+	entry->ok_pkt_offset=0;
+
+	// Find the first EOF location
+	unsigned char* it = vp;
+	it += sizeof(mysql_hdr);
+	uint64_t c_count = 0;
+	int c_count_len = mysql_decode_length(const_cast<unsigned char*>(it), &c_count);
+	it += c_count_len;
+
+	for (uint64_t i = 0; i < c_count; i++) {
+		mysql_hdr hdr;
+		memcpy(&hdr, it ,sizeof(mysql_hdr));
+		it += sizeof(mysql_hdr) + hdr.pkt_length;
+	}
+
+	if (deprecate_eof_active == false) {
+		// Store EOF position and jump to rows
+		entry->column_eof_pkt_offset = it - vp;
+		mysql_hdr hdr;
+		memcpy(&hdr, it, sizeof(mysql_hdr));
+		it += sizeof(mysql_hdr) + hdr.pkt_length;
+	}
+
+	// Find the second EOF location or the OK packet
+	for (;;) {
+		mysql_hdr hdr;
+		memcpy(&hdr, it ,sizeof(mysql_hdr));
+		unsigned char* payload = it + sizeof(mysql_hdr);
+
+		if (hdr.pkt_length < 9 && *payload == 0xfe) {
+			if (deprecate_eof_active) {
+				entry->ok_pkt_offset = it - vp;
+			} else {
+				entry->row_eof_pkt_offset = it - vp;
+			}
+			break;
+		} else {
+			it += sizeof(mysql_hdr) + hdr.pkt_length;
+		}
+	}
 
 	entry->value=(char *)malloc(vl);
 	memcpy(entry->value,vp,vl);