Merge pull request #5225 from sysown/v3.0_refactor_prepared_statement_cache_design_5211

Refactored Prepared-Statement Cache Design (Lock-Free Hot Path) - Part 2

Author: renecannao

deps/Makefile

@@ -306,6 +306,7 @@ postgresql/postgresql/src/interfaces/libpq/libpq.a:
 	cd postgresql/postgresql && patch -p0 < ../handle_row_data.patch
 	cd postgresql/postgresql && patch -p0 < ../fmt_err_msg.patch
 	cd postgresql/postgresql && patch -p0 < ../bind_fmt_text.patch
+	cd postgresql/postgresql && patch -p0 < ../pqsendpipelinesync.patch
 	#cd postgresql/postgresql && LD_LIBRARY_PATH="$(shell pwd)/libssl/openssl" ./configure --with-ssl=openssl --with-includes="$(shell pwd)/libssl/openssl/include/" --with-libraries="$(shell pwd)/libssl/openssl/" --without-readline --enable-debug  CFLAGS="-ggdb -O0 -fno-omit-frame-pointer" CPPFLAGS="-g -O0"
 	cd postgresql/postgresql && LD_LIBRARY_PATH="$(SSL_LDIR)" ./configure --with-ssl=openssl --with-includes="$(SSL_IDIR)" --with-libraries="$(SSL_LDIR)" --without-readline
 	cd postgresql/postgresql/src/interfaces/libpq && CC=${CC} CXX=${CXX} ${MAKE} MAKELEVEL=0

deps/postgresql/pqsendpipelinesync.patch

@@ -0,0 +1,84 @@
+diff --git src/interfaces/libpq/fe-exec.c src/interfaces/libpq/fe-exec.c
+index b833e76..51ad8d8 100644
+--- src/interfaces/libpq/fe-exec.c
++++ src/interfaces/libpq/fe-exec.c
+@@ -4558,3 +4558,65 @@ int PShandleRowData(PGconn *conn, bool is_first_packet, PSresult* result) {
+     return psHandleRowData(conn, is_first_packet, result);
+ }
+ 
++int
++PQsendPipelineSync(PGconn *conn)
++{
++	PGcmdQueueEntry *entry;
++
++	if (!conn)
++		return 0;
++
++	if (conn->pipelineStatus == PQ_PIPELINE_OFF)
++	{
++		libpq_append_conn_error(conn, "cannot send pipeline when not in pipeline mode");
++		return 0;
++	}
++
++	switch (conn->asyncStatus)
++	{
++		case PGASYNC_COPY_IN:
++		case PGASYNC_COPY_OUT:
++		case PGASYNC_COPY_BOTH:
++			/* should be unreachable */
++			appendPQExpBufferStr(&conn->errorMessage,
++								 "internal error: cannot send pipeline while in COPY\n");
++			return 0;
++		case PGASYNC_READY:
++		case PGASYNC_READY_MORE:
++		case PGASYNC_BUSY:
++		case PGASYNC_IDLE:
++		case PGASYNC_PIPELINE_IDLE:
++			/* OK to send sync */
++			break;
++	}
++
++	entry = pqAllocCmdQueueEntry(conn);
++	if (entry == NULL)
++		return 0;				/* error msg already set */
++
++	entry->queryclass = PGQUERY_SYNC;
++	entry->query = NULL;
++
++	/* construct the Sync message */
++	if (pqPutMsgStart('S', conn) < 0 ||
++		pqPutMsgEnd(conn) < 0)
++		goto sendFailed;
++
++	/*
++	 * Give the data a push (in pipeline mode, only if we're past the size
++	 * threshold).  In nonblock mode, don't complain if we're unable to send
++	 * it all; PQgetResult() will do any additional flushing needed.
++	 */
++	if (pqPipelineFlush(conn) < 0)
++		goto sendFailed;
++
++	/* OK, it's launched! */
++	pqAppendCmdQueueEntry(conn, entry);
++
++	return 1;
++
++sendFailed:
++	pqRecycleCmdQueueEntry(conn, entry);
++	/* error message should be set up already */
++	return 0;
++}
+diff --git src/interfaces/libpq/libpq-fe.h src/interfaces/libpq/libpq-fe.h
+index 47f25e0..b769b64 100644
+--- src/interfaces/libpq/libpq-fe.h
++++ src/interfaces/libpq/libpq-fe.h
+@@ -688,6 +688,9 @@ extern const PGresult *PQgetResultFromPGconn(PGconn *conn);
+ /* ProxySQL special handler function */
+ extern int PShandleRowData(PGconn *conn, bool is_first_packet, PSresult* result);
+ 
++/* Send a pipeline sync message without flushing the send buffer */
++extern int	PQsendPipelineSync(PGconn *conn);
++
+ #ifdef __cplusplus
+ }
+ #endif

include/PgSQL_Connection.h

@@ -12,7 +12,7 @@
 
 class PgSQL_SrvC;
 class PgSQL_Query_Result;
-class PgSQL_STMTs_local_v14;
+class PgSQL_STMT_Local;
 //class PgSQL_Describe_Prepared_Info;
 class PgSQL_Bind_Info;
 //#define STATUS_PGSQL_CONNECTION_SEQUENCE			 0x00000001
@@ -639,7 +639,7 @@ class PgSQL_Connection {
 	bool exit_pipeline_mode; // true if it is safe to exit pipeline mode
 	bool resync_failed; // true if the last resync attempt failed
 
-	PgSQL_STMTs_local_v14* local_stmts;
+	PgSQL_STMT_Local* local_stmts;
 	PgSQL_SrvC *parent;
 	PgSQL_Connection_userinfo* userinfo;
 	PgSQL_Data_Stream* myds;

include/PgSQL_ExplicitTxnStateMgr.h

@@ -56,6 +56,9 @@ struct TxnCmd {
  */
 class PgSQL_TxnCmdParser {
 public:
+    PgSQL_TxnCmdParser() noexcept { tokens.reserve(16); }
+    ~PgSQL_TxnCmdParser() noexcept = default;
+
     TxnCmd parse(std::string_view input, bool in_transaction_mode) noexcept;
 
 private:
@@ -67,14 +70,20 @@ class PgSQL_TxnCmdParser {
     TxnCmd parse_start(size_t& pos) noexcept;
 
     // Helpers
-    static std::string to_lower(std::string_view s) noexcept {
-		std::string s_copy(s);
-        std::transform(s_copy.begin(), s_copy.end(), s_copy.begin(), ::tolower);
-        return s_copy;
+    inline static bool iequals(std::string_view a, std::string_view b) noexcept {
+        if (a.size() != b.size()) return false;
+        for (size_t i = 0; i < a.size(); ++i) {
+            char ca = a[i];
+            char cb = b[i];
+            if (ca >= 'A' && ca <= 'Z') ca += 32;
+            if (cb >= 'A' && cb <= 'Z') cb += 32;
+            if (ca != cb) return false;
+        }
+        return true;
     }
 
     inline static bool contains(std::vector<std::string_view>&& list, std::string_view value) noexcept {
-        for (const auto& item : list) if (item == value) return true;
+        for (const auto& item : list) if (iequals(item, value)) return true;
         return false;
     }
 };