Resolve three scenarios that may cause workload to fail

Author: qinjingyuan

src/backend/commands/prepare.c

@@ -35,6 +35,7 @@
 #include "utils/builtins.h"
 #include "utils/snapmgr.h"
 #include "utils/timestamp.h"
+#include "utils/syscache.h"
 
 /* POLAR: Shared Server */
 #include "storage/proc.h"
@@ -51,7 +52,8 @@ static void InitQueryHashTable(void);
 static ParamListInfo EvaluateParams(PreparedStatement *pstmt, List *params,
 			   const char *queryString, EState *estate);
 static Datum build_regtype_array(Oid *param_types, int num_params);
-
+// store prepare params info
+char current_prepared_params_string[MAX_PREPARED_PARAMS_LEN];
 /*
  * Implements the 'PREPARE' utility statement.
  */
@@ -226,6 +228,29 @@ ExecuteQuery(ExecuteStmt *stmt, IntoClause *intoClause,
 	/* Look it up in the hash table */
 	entry = FetchPreparedStatement(stmt->name, true);
 
+	// get prepare params for solving parameterization problem 
+	if(entry){
+		StringInfoData current_prepared_params;
+		Oid* oid_list ;
+
+		initStringInfo(&current_prepared_params);
+		oid_list = entry->plansource->param_types;
+		appendStringInfoString(&current_prepared_params,entry->stmt_name);
+		for(int i=0;i<entry->plansource->num_params;i++){
+			Type tmp = typeidType(oid_list[i]);
+			char * typname = typeTypeName(tmp);
+			ReleaseSysCache(tmp);
+			appendStringInfoChar(&current_prepared_params,',');
+			appendStringInfoString(&current_prepared_params,typname);
+		}
+		if(current_prepared_params.len < MAX_PREPARED_PARAMS_LEN)
+			memcpy(current_prepared_params_string,current_prepared_params.data,current_prepared_params.len);
+		else 
+			memcpy(current_prepared_params_string,current_prepared_params.data,MAX_PREPARED_PARAMS_LEN);
+	}else{
+		current_prepared_params_string[0] = '\0';
+	}
+
 	/* Shouldn't find a non-fixed-result cached plan */
 	if (!entry->plansource->fixed_result)
 		elog(ERROR, "EXECUTE does not support variable-result cached plans");

src/backend/utils/error/elog.c

@@ -77,6 +77,8 @@
 #include "utils/guc.h"
 #include "utils/memutils.h"
 #include "utils/ps_status.h"
+#include "catalog/namespace.h"
+#include "commands/prepare.h"
 
 /* POLAR */
 #include "utils/polar_backtrace.h"
@@ -116,8 +118,9 @@ char	   *Log_destination_string = NULL;
 bool		syslog_sequence_numbers = true;
 bool		syslog_split_messages = true;
 /* POLAR */
-#define LOG_CHANNEL_WRITE_BUFFER_SIZE 128 * 1024 /* 128k */
+#define LOG_CHANNEL_WRITE_BUFFER_SIZE 64 * 1024 /* 64k */
 int polar_auditlog_max_query_length = POLAR_DEFAULT_MAX_AUDIT_LOG_LEN;
+bool polar_auditlog_max_query_length_limit = true;
 int polar_audit_log_flush_timeout = 0;
 /* POLAR end */
 
@@ -2603,6 +2606,19 @@ log_line_prefix(StringInfo buf, ErrorData *edata)
 					appendStringInfoString(buf, formatted_log_time);
 				break;
 			/* POLAR: new format for polar */
+			case 'L':
+				if (padding != 0)
+					appendStringInfo(buf, "%*s", padding, current_prepared_params_string);
+				else
+					appendStringInfoString(buf, current_prepared_params_string);
+				current_prepared_params_string[0] = '\0';
+				break;
+			case 'N':
+				if (padding != 0)
+					appendStringInfo(buf, "%*s", padding, namespace_search_path);
+				else
+					appendStringInfoString(buf, namespace_search_path);
+				break;
 			case 'S':
 				if (edata->is_audit_log || edata->is_slow_log)
 				{
@@ -4246,7 +4262,8 @@ polar_construct_logdata_postprocess(StringInfoData *logbuf, ErrorData *edata)
 	else
 	{
 		// NOTE(wormhole.gl): make sure logbuf length is not larger than PIPE_MAX_PAYLOAD
-		polar_shrink_audit_log(logbuf, 0);
+		if (polar_auditlog_max_query_length_limit)
+			polar_shrink_audit_log(logbuf, 0);
 	}
 
 	appendStringInfoChar(logbuf, '\n');
@@ -4287,6 +4304,8 @@ polar_write_audit_log(ErrorData *edata, const char *fmt, ...)
 	int dest;
 	char *logbuf_base = NULL;
 	int logbuf_len = 0;
+	int residual_buf_len = 0;
+	int writed_buf_len = 0;
 
 	StringInfoData logbuf;
 
@@ -4333,6 +4352,28 @@ polar_write_audit_log(ErrorData *edata, const char *fmt, ...)
 								log_channel_write_buffer_pos);
 			log_channel_write_buffer_pos = 0;
 			polar_last_audit_log_flush_time = GetCurrentTimestamp();
+			// solving incomplete audit sql log problem
+			writed_buf_len = logbuf_len - PIPE_HEADER_SIZE;
+			residual_buf_len = logbuf.len - writed_buf_len;
+			while (residual_buf_len > 0)
+			{
+				logbuf_base = log_channel_write_buffer + log_channel_write_buffer_pos;
+				logbuf_len = LOGBUF_MIN_LEN(residual_buf_len, LOG_CHANNEL_WRITE_BUFFER_SIZE - PIPE_HEADER_SIZE - log_channel_write_buffer_pos);
+				memcpy(logbuf_base + PIPE_HEADER_SIZE, logbuf.data + writed_buf_len, logbuf_len);
+				// build pipechunk header
+				dest = polar_log_dest(edata);
+				polar_construct_pipechunk_header((PipeProtoChunk *)logbuf_base, logbuf_len, dest);
+				writed_buf_len += logbuf_len;
+				residual_buf_len = residual_buf_len - logbuf_len;
+
+				logbuf_len = logbuf_len + PIPE_HEADER_SIZE;
+				log_channel_write_buffer_pos += logbuf_len;
+				polar_write_channel((PipeProtoChunk *)log_channel_write_buffer,
+									log_channel_write_buffer_pos);
+				log_channel_write_buffer_pos = 0;
+				polar_last_audit_log_flush_time = GetCurrentTimestamp();
+			}
+
 			break;
 		case LOG_DESTINATION_POLAR_SLOWLOG:
 		default:

src/backend/utils/misc/guc.c

@@ -1402,6 +1402,16 @@ static bool update_session_external_guc_index(struct config_generic *var);
 
 static struct config_bool ConfigureNamesBool[] =
 {
+	{
+		{"polar_auditlog_max_query_length_limit",PGC_SIGHUP,LOGGING,
+			gettext_noop("whether to limit polar_auditlog_max_query_length"),
+			NULL
+		},
+		&polar_auditlog_max_query_length_limit,
+		true,
+		NULL,NULL,NULL
+	},
+
 	{
 		{"polar_enable_flashback_drop",PGC_SUSET,UNGROUPED,
 			gettext_noop("whether to open flashback_drop"),

src/include/commands/prepare.h

@@ -12,6 +12,7 @@
  */
 #ifndef PREPARE_H
 #define PREPARE_H
+#define MAX_PREPARED_PARAMS_LEN 200
 
 #include "commands/explain.h"
 #include "datatype/timestamp.h"
@@ -56,5 +57,5 @@ extern TupleDesc FetchPreparedStatementResultDesc(PreparedStatement *stmt);
 extern List *FetchPreparedStatementTargetList(PreparedStatement *stmt);
 
 extern void DropAllPreparedStatements(void);
-
+extern char current_prepared_params_string[MAX_PREPARED_PARAMS_LEN];
 #endif							/* PREPARE_H */

src/include/utils/guc.h

@@ -396,6 +396,7 @@ extern bool		polar_enable_pread;
 extern bool		polar_enable_parallel_replay_standby_mode;
 extern bool		polar_enable_fallocate_walfile;
 extern bool		polar_skip_fill_walfile_zero_page;
+extern bool		polar_auditlog_max_query_length_limit;
 extern int 		polar_auditlog_max_query_length;
 extern int		polar_audit_log_flush_timeout;
 extern int 		polar_clog_max_local_cache_segments;