Display individual query in pg_stat_activity

 doc/src/sgml/config.sgml                 |   2 +-
 doc/src/sgml/monitoring.sgml             |  14 ++++
 src/backend/access/heap/vacuumlazy.c     |   2 +-
 src/backend/access/nbtree/nbtsort.c      |   2 +-
 src/backend/catalog/system_views.sql     |   1 +
 src/backend/executor/execParallel.c      |  24 ++++++-
 src/backend/postmaster/autovacuum.c      |   4 +-
 src/backend/postmaster/pgstat.c          |  57 +++++++++++++++-
 src/backend/replication/logical/worker.c |   6 +-
 src/backend/replication/walsender.c      |   6 +-
 src/backend/tcop/postgres.c              |  58 ++++++++++++----
 src/backend/utils/adt/pgstatfuncs.c      | 114 +++++++++++++++++--------------
 src/include/catalog/pg_proc.dat          |   6 +-
 src/include/pgstat.h                     |  11 ++-
 src/test/modules/worker_spi/worker_spi.c |   8 +--
 src/test/regress/expected/rules.out      |   9 +--
 16 files changed, 230 insertions(+), 94 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 7a7177c550..d2b415629b 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -7230,7 +7230,7 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv;
        <para>
        Specifies the amount of memory reserved to store the text of the
        currently executing command for each active session, for the
-       <structname>pg_stat_activity</structname>.<structfield>query</structfield> field.
+       <structname>pg_stat_activity</structname>.<structfield>query</structfield> and <structfield>individual_query</structfield> fields.
        If this value is specified without units, it is taken as bytes.
        The default value is 1024 bytes.
        This parameter can only be set at server start.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 7dcddf478a..71782e9ffa 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -896,6 +896,20 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>individual_query</structfield> <type>text</type>
+      </para>
+      <para>
+       Text of this backend's most recent individual query in case <structfield>query</structfield> contains multiple statements. If
+       <structfield>state</structfield> is <literal>active</literal> this field shows the
+       currently executing individual query. In all other states, it shows the last individual query
+       that was executed. By default the individual query text is truncated at 1024
+       bytes; this value can be changed via the parameter
+       <xref linkend="guc-track-activity-query-size"/>.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>backend_type</structfield> <type>text</type>
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 1bbc4598f7..f9a0ca4bdf 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -3489,7 +3489,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	/* Set debug_query_string for individual workers */
 	sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, false);
 	debug_query_string = sharedquery;
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, NULL);
 
 	/*
 	 * Open table.  The lock mode is the same as the leader process.  It's
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index efee86784b..11494a2f7d 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1810,7 +1810,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
 	debug_query_string = sharedquery;
 
 	/* Report the query string from leader */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, NULL);
 
 	/* Look up nbtree shared state */
 	btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8625cbeab6..a411f8b548 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -759,6 +759,7 @@ CREATE VIEW pg_stat_activity AS
             S.backend_xid,
             s.backend_xmin,
             S.query,
+            S.individual_query,
             S.backend_type
     FROM pg_stat_get_activity(NULL) AS S
         LEFT JOIN pg_database AS D ON (S.datid = D.oid)
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 382e78fb7f..19ebaf13b3 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -65,6 +65,7 @@
 #define PARALLEL_KEY_QUERY_TEXT		UINT64CONST(0xE000000000000008)
 #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
 #define PARALLEL_KEY_WAL_USAGE			UINT64CONST(0xE00000000000000A)
+#define PARALLEL_KEY_INDIVIDUAL_QUERY_TEXT		UINT64CONST(0xE00000000000000B)
 
 #define PARALLEL_TUPLE_QUEUE_SIZE		65536
 
@@ -600,7 +601,9 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	int			instrument_offset = 0;
 	Size		dsa_minsize = dsa_minimum_size();
 	char	   *query_string;
+	char	   *individual_query_string;
 	int			query_len;
+	int			individual_query_len;
 
 	/*
 	 * Force any initplan outputs that we're going to pass to workers to be
@@ -638,6 +641,15 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 						   sizeof(FixedParallelExecutorState));
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
+	/* Estimate space for individual query text. */
+	if (estate->es_plannedstmt->stmt_len == 0)
+		individual_query_len = strlen(estate->es_sourceText) - estate->es_plannedstmt->stmt_location;
+	else
+		individual_query_len = estate->es_plannedstmt->stmt_len;
+
+	shm_toc_estimate_chunk(&pcxt->estimator, individual_query_len + 1);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
 	/* Estimate space for query text. */
 	query_len = strlen(estate->es_sourceText);
 	shm_toc_estimate_chunk(&pcxt->estimator, query_len + 1);
@@ -732,6 +744,11 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	fpes->jit_flags = estate->es_jit_flags;
 	shm_toc_insert(pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fpes);
 
+	/* Store individual query string */
+	individual_query_string = shm_toc_allocate(pcxt->toc, individual_query_len + 1);
+	memcpy(individual_query_string, estate->es_sourceText + estate->es_plannedstmt->stmt_location, individual_query_len + 1);
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_INDIVIDUAL_QUERY_TEXT, individual_query_string);
+
 	/* Store query string */
 	query_string = shm_toc_allocate(pcxt->toc, query_len + 1);
 	memcpy(query_string, estate->es_sourceText, query_len + 1);
@@ -1388,6 +1405,7 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 	void	   *area_space;
 	dsa_area   *area;
 	ParallelWorkerContext pwcxt;
+	char *individual_query;
 
 	/* Get fixed-size state. */
 	fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
@@ -1403,9 +1421,13 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 
 	/* Setting debug_query_string for individual workers */
 	debug_query_string = queryDesc->sourceText;
+	individual_query = shm_toc_lookup(toc, PARALLEL_KEY_INDIVIDUAL_QUERY_TEXT, false);
 
 	/* Report workers' query for monitoring purposes */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	if (strlen(debug_query_string) != strlen(individual_query))
+		pgstat_report_activity(STATE_RUNNING, debug_query_string, individual_query);
+	else
+		pgstat_report_activity(STATE_RUNNING, debug_query_string, NULL);
 
 	/* Attach to the dynamic shared memory area. */
 	area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 9c7d4b0c60..340ab8a76e 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -3176,7 +3176,7 @@ autovac_report_activity(autovac_table *tab)
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, NULL);
 }
 
 /*
@@ -3215,7 +3215,7 @@ autovac_report_workitem(AutoVacuumWorkItem *workitem,
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, NULL);
 }
 
 /*
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 15f92b66c6..44fb7e77eb 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -2702,7 +2702,9 @@ static PgBackendStatus *MyBEEntry = NULL;
 static char *BackendAppnameBuffer = NULL;
 static char *BackendClientHostnameBuffer = NULL;
 static char *BackendActivityBuffer = NULL;
+static char *BackendIndividualActivityBuffer = NULL;
 static Size BackendActivityBufferSize = 0;
+static Size BackendIndividualActivityBufferSize = 0;
 #ifdef USE_SSL
 static PgBackendSSLStatus *BackendSslStatusBuffer = NULL;
 #endif
@@ -2730,6 +2732,9 @@ BackendStatusShmemSize(void)
 	/* BackendActivityBuffer: */
 	size = add_size(size,
 					mul_size(pgstat_track_activity_query_size, NumBackendStatSlots));
+	/* BackendIndividualActivityBuffer: */
+	size = add_size(size,
+					mul_size(pgstat_track_activity_query_size, NumBackendStatSlots));
 #ifdef USE_SSL
 	/* BackendSslStatusBuffer: */
 	size = add_size(size,
@@ -2820,6 +2825,27 @@ CreateSharedBackendStatus(void)
 		}
 	}
 
+	/* Create or attach to the indiviudal shared activity buffer */
+	BackendIndividualActivityBufferSize = mul_size(pgstat_track_activity_query_size,
+										 NumBackendStatSlots);
+	BackendIndividualActivityBuffer = (char *)
+		ShmemInitStruct("Backend Individual Activity Buffer",
+						BackendIndividualActivityBufferSize,
+						&found);
+
+	if (!found)
+	{
+		MemSet(BackendIndividualActivityBuffer, 0, BackendIndividualActivityBufferSize);
+
+		/* Initialize st_individual_activity pointers. */
+		buffer = BackendIndividualActivityBuffer;
+		for (i = 0; i < NumBackendStatSlots; i++)
+		{
+			BackendStatusArray[i].st_individual_activity_raw = buffer;
+			buffer += pgstat_track_activity_query_size;
+		}
+	}
+
 #ifdef USE_SSL
 	/* Create or attach to the shared SSL status buffer */
 	size = mul_size(sizeof(PgBackendSSLStatus), NumBackendStatSlots);
@@ -3062,10 +3088,12 @@ pgstat_bestart(void)
 	else
 		lbeentry.st_clienthostname[0] = '\0';
 	lbeentry.st_activity_raw[0] = '\0';
+	lbeentry.st_individual_activity_raw[0] = '\0';
 	/* Also make sure the last byte in each string area is always 0 */
 	lbeentry.st_appname[NAMEDATALEN - 1] = '\0';
 	lbeentry.st_clienthostname[NAMEDATALEN - 1] = '\0';
 	lbeentry.st_activity_raw[pgstat_track_activity_query_size - 1] = '\0';
+	lbeentry.st_individual_activity_raw[pgstat_track_activity_query_size - 1] = '\0';
 
 #ifdef USE_SSL
 	memcpy(lbeentry.st_sslstatus, &lsslstatus, sizeof(PgBackendSSLStatus));
@@ -3129,12 +3157,13 @@ pgstat_beshutdown_hook(int code, Datum arg)
  * ----------
  */
 void
-pgstat_report_activity(BackendState state, const char *cmd_str)
+pgstat_report_activity(BackendState state, const char *cmd_str, const char *individual_cmd_str)
 {
 	volatile PgBackendStatus *beentry = MyBEEntry;
 	TimestampTz start_timestamp;
 	TimestampTz current_timestamp;
 	int			len = 0;
+	int			individual_len = 0;
 
 	TRACE_POSTGRESQL_STATEMENT_STATUS(cmd_str);
 
@@ -3156,6 +3185,7 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 			beentry->st_state = STATE_DISABLED;
 			beentry->st_state_start_timestamp = 0;
 			beentry->st_activity_raw[0] = '\0';
+			beentry->st_individual_activity_raw[0] = '\0';
 			beentry->st_activity_start_timestamp = 0;
 			/* st_xact_start_timestamp and wait_event_info are also disabled */
 			beentry->st_xact_start_timestamp = 0;
@@ -3179,6 +3209,16 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 		 */
 		len = Min(strlen(cmd_str), pgstat_track_activity_query_size - 1);
 	}
+
+	if (individual_cmd_str != NULL)
+	{
+		/*
+		 * Compute length of to-be-stored string unaware of multi-byte
+		 * characters. For speed reasons that'll get corrected on read, rather
+		 * than computed every write.
+		 */
+		individual_len = Min(strlen(individual_cmd_str), pgstat_track_activity_query_size - 1);
+	}
 	current_timestamp = GetCurrentTimestamp();
 
 	/*
@@ -3196,6 +3236,12 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 		beentry->st_activity_start_timestamp = start_timestamp;
 	}
 
+	if (individual_cmd_str != NULL)
+	{
+		memcpy((char *) beentry->st_individual_activity_raw, individual_cmd_str, individual_len);
+		beentry->st_individual_activity_raw[len] = '\0';
+	}
+
 	PGSTAT_END_WRITE_ACTIVITY(beentry);
 }
 
@@ -3365,7 +3411,8 @@ pgstat_read_current_status(void)
 	LocalPgBackendStatus *localentry;
 	char	   *localappname,
 			   *localclienthostname,
-			   *localactivity;
+			   *localactivity,
+			   *local_individual_activity;
 #ifdef USE_SSL
 	PgBackendSSLStatus *localsslstatus;
 #endif
@@ -3400,6 +3447,9 @@ pgstat_read_current_status(void)
 	localactivity = (char *)
 		MemoryContextAllocHuge(pgStatLocalContext,
 							   pgstat_track_activity_query_size * NumBackendStatSlots);
+	local_individual_activity = (char *)
+		MemoryContextAllocHuge(pgStatLocalContext,
+							   pgstat_track_activity_query_size * NumBackendStatSlots);
 #ifdef USE_SSL
 	localsslstatus = (PgBackendSSLStatus *)
 		MemoryContextAlloc(pgStatLocalContext,
@@ -3451,6 +3501,8 @@ pgstat_read_current_status(void)
 				localentry->backendStatus.st_clienthostname = localclienthostname;
 				strcpy(localactivity, (char *) beentry->st_activity_raw);
 				localentry->backendStatus.st_activity_raw = localactivity;
+				strcpy(local_individual_activity, (char *) beentry->st_individual_activity_raw);
+				localentry->backendStatus.st_individual_activity_raw = local_individual_activity;
 #ifdef USE_SSL
 				if (beentry->st_ssl)
 				{
@@ -3489,6 +3541,7 @@ pgstat_read_current_status(void)
 			localappname += NAMEDATALEN;
 			localclienthostname += NAMEDATALEN;
 			localactivity += pgstat_track_activity_query_size;
+			local_individual_activity += pgstat_track_activity_query_size;
 #ifdef USE_SSL
 			localsslstatus++;
 #endif
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 2fcf2e61bc..de7f0ce0af 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -556,7 +556,7 @@ apply_handle_begin(StringInfo s)
 
 	in_remote_transaction = true;
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, NULL);
 }
 
 /*
@@ -600,7 +600,7 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, NULL);
 }
 
 /*
@@ -1571,7 +1571,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 												ALLOCSET_DEFAULT_SIZES);
 
 	/* mark as idle, before starting to loop */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, NULL);
 
 	for (;;)
 	{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5e2210dd7b..fd9225a987 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1601,7 +1601,7 @@ exec_replication_command(const char *cmd_string)
 	initStringInfo(&tmpbuf);
 
 	/* Report to pgstat that this process is running */
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, NULL);
 
 	switch (cmd_node->type)
 	{
@@ -1660,7 +1660,7 @@ exec_replication_command(const char *cmd_string)
 						(errmsg("cannot execute SQL commands in WAL sender for physical replication")));
 
 			/* Report to pgstat that this process is now idle */
-			pgstat_report_activity(STATE_IDLE, NULL);
+			pgstat_report_activity(STATE_IDLE, NULL, NULL);
 
 			/* Tell the caller that this wasn't a WalSender command. */
 			return false;
@@ -1679,7 +1679,7 @@ exec_replication_command(const char *cmd_string)
 	EndCommand(&qc, DestRemote, true);
 
 	/* Report to pgstat that this process is now idle */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, NULL);
 
 	return true;
 }
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index c9424f167c..98544439b0 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -167,6 +167,9 @@ static ProcSignalReason RecoveryConflictReason;
 static MemoryContext row_description_context = NULL;
 static StringInfoData row_description_buf;
 
+/* reused buffer to pass the individual queries */
+static StringInfoData individual_query_buf;
+
 /* ----------------------------------------------------------------
  *		decls for routines only used in this file
  * ----------------------------------------------------------------
@@ -991,14 +994,6 @@ exec_simple_query(const char *query_string)
 	bool		use_implicit_block;
 	char		msec_str[32];
 
-	/*
-	 * Report query to various monitoring facilities.
-	 */
-	debug_query_string = query_string;
-
-	pgstat_report_activity(STATE_RUNNING, query_string);
-
-	TRACE_POSTGRESQL_QUERY_START(query_string);
 
 	/*
 	 * We use save_log_statement_stats so ShowUsage doesn't report incorrect
@@ -1075,7 +1070,38 @@ exec_simple_query(const char *query_string)
 		Portal		portal;
 		DestReceiver *receiver;
 		int16		format;
+		char		*individual_query;
+		int 		individual_query_length;
+
+		/* if statement does not end with ;
+		 * then parsetree->stmt_len == 0
+		 */
+		if (parsetree->stmt_len ==  0)
+			individual_query_length = strlen(query_string) - parsetree->stmt_location;
+		else
+			individual_query_length = parsetree->stmt_len + 1;
 
+		/* extract the query text */
+		individual_query = palloc(individual_query_length + 1);
+		strncpy(individual_query, query_string + parsetree->stmt_location, individual_query_length);
+		individual_query[individual_query_length] = '\0';
+		/*
+		 * Report query to various monitoring facilities.
+		 */
+
+		resetStringInfo(&individual_query_buf);
+		appendStringInfoString(&individual_query_buf, individual_query);
+
+		debug_query_string = query_string;
+
+		if (strlen(debug_query_string) != individual_query_buf.len)
+			pgstat_report_activity(STATE_RUNNING, query_string, individual_query_buf.data);
+		else
+			pgstat_report_activity(STATE_RUNNING, query_string, NULL);
+
+		TRACE_POSTGRESQL_QUERY_START(query_string);
+
+		pfree(individual_query);
 		/*
 		 * Get the command name for use in status display (it also becomes the
 		 * default completion tag, down inside PortalRun).  Set ps_status and
@@ -1366,7 +1392,7 @@ exec_parse_message(const char *query_string,	/* string to execute */
 	 */
 	debug_query_string = query_string;
 
-	pgstat_report_activity(STATE_RUNNING, query_string);
+	pgstat_report_activity(STATE_RUNNING, query_string, NULL);
 
 	set_ps_display("PARSE");
 
@@ -1657,7 +1683,7 @@ exec_bind_message(StringInfo input_message)
 	 */
 	debug_query_string = psrc->query_string;
 
-	pgstat_report_activity(STATE_RUNNING, psrc->query_string);
+	pgstat_report_activity(STATE_RUNNING, psrc->query_string, NULL);
 
 	set_ps_display("BIND");
 
@@ -2115,7 +2141,7 @@ exec_execute_message(const char *portal_name, long max_rows)
 	 */
 	debug_query_string = sourceText;
 
-	pgstat_report_activity(STATE_RUNNING, sourceText);
+	pgstat_report_activity(STATE_RUNNING, sourceText, NULL);
 
 	set_ps_display(GetCommandTagName(portal->commandTag));
 
@@ -4011,6 +4037,8 @@ PostgresMain(int argc, char *argv[],
 	initStringInfo(&row_description_buf);
 	MemoryContextSwitchTo(TopMemoryContext);
 
+	initStringInfo(&individual_query_buf);
+
 	/*
 	 * Remember stand-alone backend startup time
 	 */
@@ -4196,7 +4224,7 @@ PostgresMain(int argc, char *argv[],
 			if (IsAbortedTransactionBlockState())
 			{
 				set_ps_display("idle in transaction (aborted)");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL, NULL);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4209,7 +4237,7 @@ PostgresMain(int argc, char *argv[],
 			else if (IsTransactionOrTransactionBlock())
 			{
 				set_ps_display("idle in transaction");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL, NULL);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4236,7 +4264,7 @@ PostgresMain(int argc, char *argv[],
 				pgstat_report_stat(false);
 
 				set_ps_display("idle");
-				pgstat_report_activity(STATE_IDLE, NULL);
+				pgstat_report_activity(STATE_IDLE, NULL, NULL);
 			}
 
 			ReadyForQuery(whereToSendOutput);
@@ -4384,7 +4412,7 @@ PostgresMain(int argc, char *argv[],
 				SetCurrentStatementStartTimestamp();
 
 				/* Report query to various monitoring facilities. */
-				pgstat_report_activity(STATE_FASTPATH, NULL);
+				pgstat_report_activity(STATE_FASTPATH, NULL, NULL);
 				set_ps_display("<FASTPATH>");
 
 				/* start an xact for this function invocation */
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 95738a4e34..976dfc7583 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -567,7 +567,7 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_activity(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_ACTIVITY_COLS	30
+#define PG_STAT_GET_ACTIVITY_COLS	31
 	int			num_backends = pgstat_fetch_stat_numbackends();
 	int			curr_backend;
 	int			pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -631,6 +631,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			nulls[5] = false;
 			values[5] = CStringGetTextDatum("<backend information not available>");
+			nulls[6] = false;
+			values[6] = CStringGetTextDatum("<backend information not available>");
 
 			tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 			continue;
@@ -661,20 +663,21 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[3] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xid))
-			values[15] = TransactionIdGetDatum(local_beentry->backend_xid);
+			values[16] = TransactionIdGetDatum(local_beentry->backend_xid);
 		else
-			nulls[15] = true;
+			nulls[16] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xmin))
-			values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
+			values[17] = TransactionIdGetDatum(local_beentry->backend_xmin);
 		else
-			nulls[16] = true;
+			nulls[17] = true;
 
 		/* Values only available to role member or pg_read_all_stats */
 		if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
 		{
 			SockAddr	zero_clientaddr;
 			char	   *clipped_activity;
+			char	   *clipped_individual_activity;
 
 			switch (beentry->st_state)
 			{
@@ -705,8 +708,12 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			values[5] = CStringGetTextDatum(clipped_activity);
 			pfree(clipped_activity);
 
+			clipped_individual_activity = pgstat_clip_activity(beentry->st_individual_activity_raw);
+			values[6] = CStringGetTextDatum(clipped_individual_activity);
+			pfree(clipped_individual_activity);
+
 			/* leader_pid */
-			nulls[29] = true;
+			nulls[30] = true;
 
 			proc = BackendPidGetProc(beentry->st_procpid);
 
@@ -743,20 +750,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 				 */
 				if (leader && leader->pid != beentry->st_procpid)
 				{
-					values[29] = Int32GetDatum(leader->pid);
-					nulls[29] = false;
+					values[30] = Int32GetDatum(leader->pid);
+					nulls[30] = false;
 				}
 			}
 
 			if (wait_event_type)
-				values[6] = CStringGetTextDatum(wait_event_type);
+				values[7] = CStringGetTextDatum(wait_event_type);
 			else
-				nulls[6] = true;
+				nulls[7] = true;
 
 			if (wait_event)
-				values[7] = CStringGetTextDatum(wait_event);
+				values[8] = CStringGetTextDatum(wait_event);
 			else
-				nulls[7] = true;
+				nulls[8] = true;
 
 			/*
 			 * Don't expose transaction time for walsenders; it confuses
@@ -765,33 +772,33 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			 */
 			if (beentry->st_xact_start_timestamp != 0 &&
 				beentry->st_backendType != B_WAL_SENDER)
-				values[8] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
+				values[9] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
 			else
-				nulls[8] = true;
+				nulls[9] = true;
 
 			if (beentry->st_activity_start_timestamp != 0)
-				values[9] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
+				values[10] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
 			else
-				nulls[9] = true;
+				nulls[10] = true;
 
 			if (beentry->st_proc_start_timestamp != 0)
-				values[10] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
+				values[11] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
 			else
-				nulls[10] = true;
+				nulls[11] = true;
 
 			if (beentry->st_state_start_timestamp != 0)
-				values[11] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
+				values[12] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
 			else
-				nulls[11] = true;
+				nulls[12] = true;
 
 			/* A zeroed client addr means we don't know */
 			memset(&zero_clientaddr, 0, sizeof(zero_clientaddr));
 			if (memcmp(&(beentry->st_clientaddr), &zero_clientaddr,
 					   sizeof(zero_clientaddr)) == 0)
 			{
-				nulls[12] = true;
 				nulls[13] = true;
 				nulls[14] = true;
+				nulls[15] = true;
 			}
 			else
 			{
@@ -815,20 +822,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					if (ret == 0)
 					{
 						clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);
-						values[12] = DirectFunctionCall1(inet_in,
+						values[13] = DirectFunctionCall1(inet_in,
 														 CStringGetDatum(remote_host));
 						if (beentry->st_clienthostname &&
 							beentry->st_clienthostname[0])
-							values[13] = CStringGetTextDatum(beentry->st_clienthostname);
+							values[14] = CStringGetTextDatum(beentry->st_clienthostname);
 						else
-							nulls[13] = true;
-						values[14] = Int32GetDatum(atoi(remote_port));
+							nulls[14] = true;
+						values[15] = Int32GetDatum(atoi(remote_port));
 					}
 					else
 					{
-						nulls[12] = true;
 						nulls[13] = true;
 						nulls[14] = true;
+						nulls[15] = true;
 					}
 				}
 				else if (beentry->st_clientaddr.addr.ss_family == AF_UNIX)
@@ -839,16 +846,16 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					 * connections we have no permissions to view, or with
 					 * errors.
 					 */
-					nulls[12] = true;
 					nulls[13] = true;
-					values[14] = Int32GetDatum(-1);
+					nulls[14] = true;
+					values[15] = Int32GetDatum(-1);
 				}
 				else
 				{
 					/* Unknown address type, should never happen */
-					nulls[12] = true;
 					nulls[13] = true;
 					nulls[14] = true;
+					nulls[15] = true;
 				}
 			}
 			/* Add backend type */
@@ -858,59 +865,59 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 				bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
 				if (bgw_type)
-					values[17] = CStringGetTextDatum(bgw_type);
+					values[18] = CStringGetTextDatum(bgw_type);
 				else
-					nulls[17] = true;
+					nulls[18] = true;
 			}
 			else
-				values[17] =
+				values[18] =
 					CStringGetTextDatum(GetBackendTypeDesc(beentry->st_backendType));
 
 			/* SSL information */
 			if (beentry->st_ssl)
 			{
-				values[18] = BoolGetDatum(true);	/* ssl */
-				values[19] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
-				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
-				values[21] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
-				values[22] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
+				values[19] = BoolGetDatum(true);	/* ssl */
+				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
+				values[21] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
+				values[22] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
+				values[23] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
 
 				if (beentry->st_sslstatus->ssl_client_dn[0])
-					values[23] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
+					values[24] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
 				else
-					nulls[23] = true;
+					nulls[24] = true;
 
 				if (beentry->st_sslstatus->ssl_client_serial[0])
-					values[24] = DirectFunctionCall3(numeric_in,
+					values[25] = DirectFunctionCall3(numeric_in,
 													 CStringGetDatum(beentry->st_sslstatus->ssl_client_serial),
 													 ObjectIdGetDatum(InvalidOid),
 													 Int32GetDatum(-1));
 				else
-					nulls[24] = true;
+					nulls[25] = true;
 
 				if (beentry->st_sslstatus->ssl_issuer_dn[0])
-					values[25] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
+					values[26] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
 				else
-					nulls[25] = true;
+					nulls[26] = true;
 			}
 			else
 			{
-				values[18] = BoolGetDatum(false);	/* ssl */
-				nulls[19] = nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = true;
+				values[19] = BoolGetDatum(false);	/* ssl */
+				nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = nulls[26] = true;
 			}
 
 			/* GSSAPI information */
 			if (beentry->st_gss)
 			{
-				values[26] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
-				values[27] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
-				values[28] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
+				values[27] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
+				values[28] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
+				values[29] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
 			}
 			else
 			{
-				values[26] = BoolGetDatum(false);	/* gss_auth */
-				nulls[27] = true;	/* No GSS principal */
-				values[28] = BoolGetDatum(false);	/* GSS Encryption not in
+				values[27] = BoolGetDatum(false);	/* gss_auth */
+				nulls[28] = true;	/* No GSS principal */
+				values[29] = BoolGetDatum(false);	/* GSS Encryption not in
 													 * use */
 			}
 		}
@@ -918,8 +925,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 		{
 			/* No permissions to view data about this session */
 			values[5] = CStringGetTextDatum("<insufficient privilege>");
+			values[6] = CStringGetTextDatum("<insufficient privilege>");
 			nulls[4] = true;
-			nulls[6] = true;
 			nulls[7] = true;
 			nulls[8] = true;
 			nulls[9] = true;
@@ -928,7 +935,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[12] = true;
 			nulls[13] = true;
 			nulls[14] = true;
-			nulls[17] = true;
+			nulls[15] = true;
 			nulls[18] = true;
 			nulls[19] = true;
 			nulls[20] = true;
@@ -941,6 +948,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[27] = true;
 			nulls[28] = true;
 			nulls[29] = true;
+			nulls[30] = true;
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 082a11f270..a4451c1ef6 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5222,9 +5222,9 @@
   proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'int4',
-  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
+  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,individual_query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
   prosrc => 'pg_stat_get_activity' },
 { oid => '3318',
   descr => 'statistics: information about progress of backends running maintenance command',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 1387201382..f29dfcbf13 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -1123,6 +1123,15 @@ typedef struct PgBackendStatus
 	 */
 	char	   *st_activity_raw;
 
+	/*
+	 * Current individual command string; MUST be null-terminated. Note that this string
+	 * possibly is truncated in the middle of a multi-byte character. As
+	 * activity strings are stored more frequently than read, that allows to
+	 * move the cost of correct truncation to the display side. Use
+	 * pgstat_clip_activity() to truncate correctly.
+	 */
+	char	   *st_individual_activity_raw;
+
 	/*
 	 * Command progress reporting.  Any command which wishes can advertise
 	 * that it is running by setting st_progress_command,
@@ -1314,7 +1323,7 @@ extern void pgstat_report_checksum_failure(void);
 extern void pgstat_initialize(void);
 extern void pgstat_bestart(void);
 
-extern void pgstat_report_activity(BackendState state, const char *cmd_str);
+extern void pgstat_report_activity(BackendState state, const char *cmd_str, const char *individual_cmd_str);
 extern void pgstat_report_tempfile(size_t filesize);
 extern void pgstat_report_appname(const char *appname);
 extern void pgstat_report_xact_timestamp(TimestampTz tstamp);
diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index 1c7b17c56f..bc7b4a4a54 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -112,7 +112,7 @@ initialize_worker_spi(worktable *table)
 	StartTransactionCommand();
 	SPI_connect();
 	PushActiveSnapshot(GetTransactionSnapshot());
-	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
+	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema", NULL);
 
 	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
 	initStringInfo(&buf);
@@ -156,7 +156,7 @@ initialize_worker_spi(worktable *table)
 	SPI_finish();
 	PopActiveSnapshot();
 	CommitTransactionCommand();
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, NULL);
 }
 
 void
@@ -262,7 +262,7 @@ worker_spi_main(Datum main_arg)
 		StartTransactionCommand();
 		SPI_connect();
 		PushActiveSnapshot(GetTransactionSnapshot());
-		pgstat_report_activity(STATE_RUNNING, buf.data);
+		pgstat_report_activity(STATE_RUNNING, buf.data, NULL);
 
 		/* We can now execute queries via SPI */
 		ret = SPI_execute(buf.data, false, 0);
@@ -292,7 +292,7 @@ worker_spi_main(Datum main_arg)
 		PopActiveSnapshot();
 		CommitTransactionCommand();
 		pgstat_report_stat(false);
-		pgstat_report_activity(STATE_IDLE, NULL);
+		pgstat_report_activity(STATE_IDLE, NULL, NULL);
 	}
 
 	proc_exit(1);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 601734a6f1..5635e80811 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1751,8 +1751,9 @@ pg_stat_activity| SELECT s.datid,
     s.backend_xid,
     s.backend_xmin,
     s.query,
+    s.individual_query,
     s.backend_type
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1857,7 +1858,7 @@ pg_stat_gssapi| SELECT s.pid,
     s.gss_auth AS gss_authenticated,
     s.gss_princ AS principal,
     s.gss_enc AS encrypted
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_progress_analyze| SELECT s.pid,
     s.datid,
@@ -2005,7 +2006,7 @@ pg_stat_replication| SELECT s.pid,
     w.sync_priority,
     w.sync_state,
     w.reply_time
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_slru| SELECT s.name,
@@ -2027,7 +2028,7 @@ pg_stat_ssl| SELECT s.pid,
     s.ssl_client_dn AS client_dn,
     s.ssl_client_serial AS client_serial,
     s.ssl_issuer_dn AS issuer_dn
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,

 doc/src/sgml/config.sgml                 |   2 +-
 doc/src/sgml/monitoring.sgml             |  14 ++++
 src/backend/access/heap/vacuumlazy.c     |   2 +-
 src/backend/access/nbtree/nbtsort.c      |   2 +-
 src/backend/catalog/system_views.sql     |   1 +
 src/backend/executor/execParallel.c      |  16 +++-
 src/backend/postmaster/autovacuum.c      |   4 +-
 src/backend/postmaster/pgstat.c          |   8 +-
 src/backend/replication/logical/worker.c |   6 +-
 src/backend/replication/walsender.c      |   6 +-
 src/backend/tcop/postgres.c              |  31 ++++----
 src/backend/utils/adt/pgstatfuncs.c      | 132 ++++++++++++++++++-------------
 src/include/catalog/pg_proc.dat          |   6 +-
 src/include/pgstat.h                     |  15 +++-
 src/test/modules/worker_spi/worker_spi.c |   8 +-
 src/test/regress/expected/rules.out      |   9 ++-
 16 files changed, 169 insertions(+), 93 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 7a7177c550..d2b415629b 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -7230,7 +7230,7 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv;
        <para>
        Specifies the amount of memory reserved to store the text of the
        currently executing command for each active session, for the
-       <structname>pg_stat_activity</structname>.<structfield>query</structfield> field.
+       <structname>pg_stat_activity</structname>.<structfield>query</structfield> and <structfield>individual_query</structfield> fields.
        If this value is specified without units, it is taken as bytes.
        The default value is 1024 bytes.
        This parameter can only be set at server start.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 7dcddf478a..71782e9ffa 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -896,6 +896,20 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>individual_query</structfield> <type>text</type>
+      </para>
+      <para>
+       Text of this backend's most recent individual query in case <structfield>query</structfield> contains multiple statements. If
+       <structfield>state</structfield> is <literal>active</literal> this field shows the
+       currently executing individual query. In all other states, it shows the last individual query
+       that was executed. By default the individual query text is truncated at 1024
+       bytes; this value can be changed via the parameter
+       <xref linkend="guc-track-activity-query-size"/>.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>backend_type</structfield> <type>text</type>
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 1bbc4598f7..00efbf97cb 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -3489,7 +3489,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	/* Set debug_query_string for individual workers */
 	sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, false);
 	debug_query_string = sharedquery;
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/*
 	 * Open table.  The lock mode is the same as the leader process.  It's
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index efee86784b..81725bd0ac 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1810,7 +1810,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
 	debug_query_string = sharedquery;
 
 	/* Report the query string from leader */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/* Look up nbtree shared state */
 	btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8625cbeab6..a411f8b548 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -759,6 +759,7 @@ CREATE VIEW pg_stat_activity AS
             S.backend_xid,
             s.backend_xmin,
             S.query,
+            S.individual_query,
             S.backend_type
     FROM pg_stat_get_activity(NULL) AS S
         LEFT JOIN pg_database AS D ON (S.datid = D.oid)
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 382e78fb7f..e0587d57b4 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -65,6 +65,7 @@
 #define PARALLEL_KEY_QUERY_TEXT		UINT64CONST(0xE000000000000008)
 #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
 #define PARALLEL_KEY_WAL_USAGE			UINT64CONST(0xE00000000000000A)
+#define PARALLEL_KEY_INDIVIDUAL_QUERY	UINT64CONST(0xE00000000000000B)
 
 #define PARALLEL_TUPLE_QUEUE_SIZE		65536
 
@@ -601,6 +602,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	Size		dsa_minsize = dsa_minimum_size();
 	char	   *query_string;
 	int			query_len;
+	PgStat_IndividualQuery *individual_query;
 
 	/*
 	 * Force any initplan outputs that we're going to pass to workers to be
@@ -638,6 +640,10 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 						   sizeof(FixedParallelExecutorState));
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
+	/* Estimate space for individual query. */
+	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(PgStat_IndividualQuery));
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
 	/* Estimate space for query text. */
 	query_len = strlen(estate->es_sourceText);
 	shm_toc_estimate_chunk(&pcxt->estimator, query_len + 1);
@@ -732,6 +738,12 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	fpes->jit_flags = estate->es_jit_flags;
 	shm_toc_insert(pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fpes);
 
+	/* individual query stmt_len and stmt_location*/
+	individual_query = shm_toc_allocate(pcxt->toc, sizeof(PgStat_IndividualQuery));
+	individual_query->stmt_len = estate->es_plannedstmt->stmt_len;
+	individual_query->stmt_location = estate->es_plannedstmt->stmt_location;
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_INDIVIDUAL_QUERY, individual_query);
+
 	/* Store query string */
 	query_string = shm_toc_allocate(pcxt->toc, query_len + 1);
 	memcpy(query_string, estate->es_sourceText, query_len + 1);
@@ -1388,6 +1400,7 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 	void	   *area_space;
 	dsa_area   *area;
 	ParallelWorkerContext pwcxt;
+	PgStat_IndividualQuery *individual_query;
 
 	/* Get fixed-size state. */
 	fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
@@ -1403,9 +1416,10 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 
 	/* Setting debug_query_string for individual workers */
 	debug_query_string = queryDesc->sourceText;
+	individual_query = shm_toc_lookup(toc, PARALLEL_KEY_INDIVIDUAL_QUERY, false);
 
 	/* Report workers' query for monitoring purposes */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, individual_query->stmt_location, individual_query->stmt_len);
 
 	/* Attach to the dynamic shared memory area. */
 	area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 9c7d4b0c60..4c023b6d17 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -3176,7 +3176,7 @@ autovac_report_activity(autovac_table *tab)
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
@@ -3215,7 +3215,7 @@ autovac_report_workitem(AutoVacuumWorkItem *workitem,
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 15f92b66c6..d313337e81 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -3129,7 +3129,7 @@ pgstat_beshutdown_hook(int code, Datum arg)
  * ----------
  */
 void
-pgstat_report_activity(BackendState state, const char *cmd_str)
+pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len)
 {
 	volatile PgBackendStatus *beentry = MyBEEntry;
 	TimestampTz start_timestamp;
@@ -3156,6 +3156,8 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 			beentry->st_state = STATE_DISABLED;
 			beentry->st_state_start_timestamp = 0;
 			beentry->st_activity_raw[0] = '\0';
+			beentry->stmt_location = 0;
+			beentry->stmt_len = 0;
 			beentry->st_activity_start_timestamp = 0;
 			/* st_xact_start_timestamp and wait_event_info are also disabled */
 			beentry->st_xact_start_timestamp = 0;
@@ -3178,7 +3180,11 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 		 * than computed every write.
 		 */
 		len = Min(strlen(cmd_str), pgstat_track_activity_query_size - 1);
+		/* Individual query */
+		beentry->stmt_location = stmt_location;
+		beentry->stmt_len = stmt_len;
 	}
+
 	current_timestamp = GetCurrentTimestamp();
 
 	/*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 2fcf2e61bc..a82616fb99 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -556,7 +556,7 @@ apply_handle_begin(StringInfo s)
 
 	in_remote_transaction = true;
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 }
 
 /*
@@ -600,7 +600,7 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 /*
@@ -1571,7 +1571,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 												ALLOCSET_DEFAULT_SIZES);
 
 	/* mark as idle, before starting to loop */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 	for (;;)
 	{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5e2210dd7b..2fa3bb8839 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1601,7 +1601,7 @@ exec_replication_command(const char *cmd_string)
 	initStringInfo(&tmpbuf);
 
 	/* Report to pgstat that this process is running */
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 
 	switch (cmd_node->type)
 	{
@@ -1660,7 +1660,7 @@ exec_replication_command(const char *cmd_string)
 						(errmsg("cannot execute SQL commands in WAL sender for physical replication")));
 
 			/* Report to pgstat that this process is now idle */
-			pgstat_report_activity(STATE_IDLE, NULL);
+			pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 			/* Tell the caller that this wasn't a WalSender command. */
 			return false;
@@ -1679,7 +1679,7 @@ exec_replication_command(const char *cmd_string)
 	EndCommand(&qc, DestRemote, true);
 
 	/* Report to pgstat that this process is now idle */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 	return true;
 }
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index c9424f167c..4cffc5540c 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -991,14 +991,6 @@ exec_simple_query(const char *query_string)
 	bool		use_implicit_block;
 	char		msec_str[32];
 
-	/*
-	 * Report query to various monitoring facilities.
-	 */
-	debug_query_string = query_string;
-
-	pgstat_report_activity(STATE_RUNNING, query_string);
-
-	TRACE_POSTGRESQL_QUERY_START(query_string);
 
 	/*
 	 * We use save_log_statement_stats so ShowUsage doesn't report incorrect
@@ -1076,6 +1068,15 @@ exec_simple_query(const char *query_string)
 		DestReceiver *receiver;
 		int16		format;
 
+		/*
+		 * Report query to various monitoring facilities.
+		 */
+		debug_query_string = query_string;
+
+		pgstat_report_activity(STATE_RUNNING, query_string, parsetree->stmt_location, parsetree->stmt_len);
+
+		TRACE_POSTGRESQL_QUERY_START(query_string);
+
 		/*
 		 * Get the command name for use in status display (it also becomes the
 		 * default completion tag, down inside PortalRun).  Set ps_status and
@@ -1366,7 +1367,7 @@ exec_parse_message(const char *query_string,	/* string to execute */
 	 */
 	debug_query_string = query_string;
 
-	pgstat_report_activity(STATE_RUNNING, query_string);
+	pgstat_report_activity(STATE_RUNNING, query_string, 0, 0);
 
 	set_ps_display("PARSE");
 
@@ -1657,7 +1658,7 @@ exec_bind_message(StringInfo input_message)
 	 */
 	debug_query_string = psrc->query_string;
 
-	pgstat_report_activity(STATE_RUNNING, psrc->query_string);
+	pgstat_report_activity(STATE_RUNNING, psrc->query_string, 0, 0);
 
 	set_ps_display("BIND");
 
@@ -2115,7 +2116,7 @@ exec_execute_message(const char *portal_name, long max_rows)
 	 */
 	debug_query_string = sourceText;
 
-	pgstat_report_activity(STATE_RUNNING, sourceText);
+	pgstat_report_activity(STATE_RUNNING, sourceText, 0, 0);
 
 	set_ps_display(GetCommandTagName(portal->commandTag));
 
@@ -4196,7 +4197,7 @@ PostgresMain(int argc, char *argv[],
 			if (IsAbortedTransactionBlockState())
 			{
 				set_ps_display("idle in transaction (aborted)");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4209,7 +4210,7 @@ PostgresMain(int argc, char *argv[],
 			else if (IsTransactionOrTransactionBlock())
 			{
 				set_ps_display("idle in transaction");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4236,7 +4237,7 @@ PostgresMain(int argc, char *argv[],
 				pgstat_report_stat(false);
 
 				set_ps_display("idle");
-				pgstat_report_activity(STATE_IDLE, NULL);
+				pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 			}
 
 			ReadyForQuery(whereToSendOutput);
@@ -4384,7 +4385,7 @@ PostgresMain(int argc, char *argv[],
 				SetCurrentStatementStartTimestamp();
 
 				/* Report query to various monitoring facilities. */
-				pgstat_report_activity(STATE_FASTPATH, NULL);
+				pgstat_report_activity(STATE_FASTPATH, NULL, 0, 0);
 				set_ps_display("<FASTPATH>");
 
 				/* start an xact for this function invocation */
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 95738a4e34..0700a227c6 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -567,7 +567,7 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_activity(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_ACTIVITY_COLS	30
+#define PG_STAT_GET_ACTIVITY_COLS	31
 	int			num_backends = pgstat_fetch_stat_numbackends();
 	int			curr_backend;
 	int			pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -631,6 +631,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			nulls[5] = false;
 			values[5] = CStringGetTextDatum("<backend information not available>");
+			nulls[6] = false;
+			values[6] = CStringGetTextDatum("<backend information not available>");
 
 			tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 			continue;
@@ -661,20 +663,21 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[3] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xid))
-			values[15] = TransactionIdGetDatum(local_beentry->backend_xid);
+			values[16] = TransactionIdGetDatum(local_beentry->backend_xid);
 		else
-			nulls[15] = true;
+			nulls[16] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xmin))
-			values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
+			values[17] = TransactionIdGetDatum(local_beentry->backend_xmin);
 		else
-			nulls[16] = true;
+			nulls[17] = true;
 
 		/* Values only available to role member or pg_read_all_stats */
 		if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
 		{
 			SockAddr	zero_clientaddr;
 			char	   *clipped_activity;
+			int		individual_len;
 
 			switch (beentry->st_state)
 			{
@@ -703,10 +706,32 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			clipped_activity = pgstat_clip_activity(beentry->st_activity_raw);
 			values[5] = CStringGetTextDatum(clipped_activity);
+
+			if (clipped_activity[0] != '\0')
+			{
+			/* display individual query */
+				if ((beentry->stmt_len + beentry->stmt_location < pgstat_track_activity_query_size)
+					&& (beentry->stmt_len + 1 != strlen(clipped_activity))
+					&& (beentry->stmt_len + beentry->stmt_location != 0))
+				{
+					if (beentry->stmt_len ==  0)
+						individual_len = strlen(clipped_activity) - beentry->stmt_location;
+					else
+						individual_len = beentry->stmt_len + 1;
+
+					clipped_activity[beentry->stmt_location + individual_len] = '\0';
+					values[6] = CStringGetTextDatum(clipped_activity + beentry->stmt_location);
+				} else {
+					nulls[6] = true;
+				}
+			} else {
+				nulls[6] = true;
+			}
+
 			pfree(clipped_activity);
 
 			/* leader_pid */
-			nulls[29] = true;
+			nulls[30] = true;
 
 			proc = BackendPidGetProc(beentry->st_procpid);
 
@@ -743,20 +768,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 				 */
 				if (leader && leader->pid != beentry->st_procpid)
 				{
-					values[29] = Int32GetDatum(leader->pid);
-					nulls[29] = false;
+					values[30] = Int32GetDatum(leader->pid);
+					nulls[30] = false;
 				}
 			}
 
 			if (wait_event_type)
-				values[6] = CStringGetTextDatum(wait_event_type);
+				values[7] = CStringGetTextDatum(wait_event_type);
 			else
-				nulls[6] = true;
+				nulls[7] = true;
 
 			if (wait_event)
-				values[7] = CStringGetTextDatum(wait_event);
+				values[8] = CStringGetTextDatum(wait_event);
 			else
-				nulls[7] = true;
+				nulls[8] = true;
 
 			/*
 			 * Don't expose transaction time for walsenders; it confuses
@@ -765,33 +790,33 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			 */
 			if (beentry->st_xact_start_timestamp != 0 &&
 				beentry->st_backendType != B_WAL_SENDER)
-				values[8] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
+				values[9] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
 			else
-				nulls[8] = true;
+				nulls[9] = true;
 
 			if (beentry->st_activity_start_timestamp != 0)
-				values[9] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
+				values[10] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
 			else
-				nulls[9] = true;
+				nulls[10] = true;
 
 			if (beentry->st_proc_start_timestamp != 0)
-				values[10] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
+				values[11] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
 			else
-				nulls[10] = true;
+				nulls[11] = true;
 
 			if (beentry->st_state_start_timestamp != 0)
-				values[11] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
+				values[12] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
 			else
-				nulls[11] = true;
+				nulls[12] = true;
 
 			/* A zeroed client addr means we don't know */
 			memset(&zero_clientaddr, 0, sizeof(zero_clientaddr));
 			if (memcmp(&(beentry->st_clientaddr), &zero_clientaddr,
 					   sizeof(zero_clientaddr)) == 0)
 			{
-				nulls[12] = true;
 				nulls[13] = true;
 				nulls[14] = true;
+				nulls[15] = true;
 			}
 			else
 			{
@@ -815,20 +840,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					if (ret == 0)
 					{
 						clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);
-						values[12] = DirectFunctionCall1(inet_in,
+						values[13] = DirectFunctionCall1(inet_in,
 														 CStringGetDatum(remote_host));
 						if (beentry->st_clienthostname &&
 							beentry->st_clienthostname[0])
-							values[13] = CStringGetTextDatum(beentry->st_clienthostname);
+							values[14] = CStringGetTextDatum(beentry->st_clienthostname);
 						else
-							nulls[13] = true;
-						values[14] = Int32GetDatum(atoi(remote_port));
+							nulls[14] = true;
+						values[15] = Int32GetDatum(atoi(remote_port));
 					}
 					else
 					{
-						nulls[12] = true;
 						nulls[13] = true;
 						nulls[14] = true;
+						nulls[15] = true;
 					}
 				}
 				else if (beentry->st_clientaddr.addr.ss_family == AF_UNIX)
@@ -839,16 +864,16 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					 * connections we have no permissions to view, or with
 					 * errors.
 					 */
-					nulls[12] = true;
 					nulls[13] = true;
-					values[14] = Int32GetDatum(-1);
+					nulls[14] = true;
+					values[15] = Int32GetDatum(-1);
 				}
 				else
 				{
 					/* Unknown address type, should never happen */
-					nulls[12] = true;
 					nulls[13] = true;
 					nulls[14] = true;
+					nulls[15] = true;
 				}
 			}
 			/* Add backend type */
@@ -858,59 +883,59 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 				bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
 				if (bgw_type)
-					values[17] = CStringGetTextDatum(bgw_type);
+					values[18] = CStringGetTextDatum(bgw_type);
 				else
-					nulls[17] = true;
+					nulls[18] = true;
 			}
 			else
-				values[17] =
+				values[18] =
 					CStringGetTextDatum(GetBackendTypeDesc(beentry->st_backendType));
 
 			/* SSL information */
 			if (beentry->st_ssl)
 			{
-				values[18] = BoolGetDatum(true);	/* ssl */
-				values[19] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
-				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
-				values[21] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
-				values[22] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
+				values[19] = BoolGetDatum(true);	/* ssl */
+				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
+				values[21] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
+				values[22] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
+				values[23] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
 
 				if (beentry->st_sslstatus->ssl_client_dn[0])
-					values[23] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
+					values[24] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
 				else
-					nulls[23] = true;
+					nulls[24] = true;
 
 				if (beentry->st_sslstatus->ssl_client_serial[0])
-					values[24] = DirectFunctionCall3(numeric_in,
+					values[25] = DirectFunctionCall3(numeric_in,
 													 CStringGetDatum(beentry->st_sslstatus->ssl_client_serial),
 													 ObjectIdGetDatum(InvalidOid),
 													 Int32GetDatum(-1));
 				else
-					nulls[24] = true;
+					nulls[25] = true;
 
 				if (beentry->st_sslstatus->ssl_issuer_dn[0])
-					values[25] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
+					values[26] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
 				else
-					nulls[25] = true;
+					nulls[26] = true;
 			}
 			else
 			{
-				values[18] = BoolGetDatum(false);	/* ssl */
-				nulls[19] = nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = true;
+				values[19] = BoolGetDatum(false);	/* ssl */
+				nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = nulls[26] = true;
 			}
 
 			/* GSSAPI information */
 			if (beentry->st_gss)
 			{
-				values[26] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
-				values[27] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
-				values[28] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
+				values[27] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
+				values[28] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
+				values[29] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
 			}
 			else
 			{
-				values[26] = BoolGetDatum(false);	/* gss_auth */
-				nulls[27] = true;	/* No GSS principal */
-				values[28] = BoolGetDatum(false);	/* GSS Encryption not in
+				values[27] = BoolGetDatum(false);	/* gss_auth */
+				nulls[28] = true;	/* No GSS principal */
+				values[29] = BoolGetDatum(false);	/* GSS Encryption not in
 													 * use */
 			}
 		}
@@ -918,8 +943,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 		{
 			/* No permissions to view data about this session */
 			values[5] = CStringGetTextDatum("<insufficient privilege>");
+			values[6] = CStringGetTextDatum("<insufficient privilege>");
 			nulls[4] = true;
-			nulls[6] = true;
 			nulls[7] = true;
 			nulls[8] = true;
 			nulls[9] = true;
@@ -928,7 +953,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[12] = true;
 			nulls[13] = true;
 			nulls[14] = true;
-			nulls[17] = true;
+			nulls[15] = true;
 			nulls[18] = true;
 			nulls[19] = true;
 			nulls[20] = true;
@@ -941,6 +966,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[27] = true;
 			nulls[28] = true;
 			nulls[29] = true;
+			nulls[30] = true;
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 082a11f270..a4451c1ef6 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5222,9 +5222,9 @@
   proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'int4',
-  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
+  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,individual_query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
   prosrc => 'pg_stat_get_activity' },
 { oid => '3318',
   descr => 'statistics: information about progress of backends running maintenance command',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 1387201382..df245aa042 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -118,6 +118,13 @@ typedef struct PgStat_TableCounts
 	PgStat_Counter t_blocks_hit;
 } PgStat_TableCounts;
 
+/* Individual Query location and length*/
+typedef struct PgStat_IndividualQuery
+{
+	int         stmt_location;
+	int         stmt_len;
+} PgStat_IndividualQuery;
+
 /* Possible targets for resetting cluster-wide shared values */
 typedef enum PgStat_Shared_Reset_Target
 {
@@ -1123,6 +1130,12 @@ typedef struct PgBackendStatus
 	 */
 	char	   *st_activity_raw;
 
+	/*
+	 * Current individual command location and length
+	 */
+	int stmt_location;
+	int stmt_len;
+
 	/*
 	 * Command progress reporting.  Any command which wishes can advertise
 	 * that it is running by setting st_progress_command,
@@ -1314,7 +1327,7 @@ extern void pgstat_report_checksum_failure(void);
 extern void pgstat_initialize(void);
 extern void pgstat_bestart(void);
 
-extern void pgstat_report_activity(BackendState state, const char *cmd_str);
+extern void pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len);
 extern void pgstat_report_tempfile(size_t filesize);
 extern void pgstat_report_appname(const char *appname);
 extern void pgstat_report_xact_timestamp(TimestampTz tstamp);
diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index 1c7b17c56f..c4d528f291 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -112,7 +112,7 @@ initialize_worker_spi(worktable *table)
 	StartTransactionCommand();
 	SPI_connect();
 	PushActiveSnapshot(GetTransactionSnapshot());
-	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
+	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema", 0, 0);
 
 	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
 	initStringInfo(&buf);
@@ -156,7 +156,7 @@ initialize_worker_spi(worktable *table)
 	SPI_finish();
 	PopActiveSnapshot();
 	CommitTransactionCommand();
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 void
@@ -262,7 +262,7 @@ worker_spi_main(Datum main_arg)
 		StartTransactionCommand();
 		SPI_connect();
 		PushActiveSnapshot(GetTransactionSnapshot());
-		pgstat_report_activity(STATE_RUNNING, buf.data);
+		pgstat_report_activity(STATE_RUNNING, buf.data, 0, 0);
 
 		/* We can now execute queries via SPI */
 		ret = SPI_execute(buf.data, false, 0);
@@ -292,7 +292,7 @@ worker_spi_main(Datum main_arg)
 		PopActiveSnapshot();
 		CommitTransactionCommand();
 		pgstat_report_stat(false);
-		pgstat_report_activity(STATE_IDLE, NULL);
+		pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 	}
 
 	proc_exit(1);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 601734a6f1..5635e80811 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1751,8 +1751,9 @@ pg_stat_activity| SELECT s.datid,
     s.backend_xid,
     s.backend_xmin,
     s.query,
+    s.individual_query,
     s.backend_type
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1857,7 +1858,7 @@ pg_stat_gssapi| SELECT s.pid,
     s.gss_auth AS gss_authenticated,
     s.gss_princ AS principal,
     s.gss_enc AS encrypted
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_progress_analyze| SELECT s.pid,
     s.datid,
@@ -2005,7 +2006,7 @@ pg_stat_replication| SELECT s.pid,
     w.sync_priority,
     w.sync_state,
     w.reply_time
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_slru| SELECT s.name,
@@ -2027,7 +2028,7 @@ pg_stat_ssl| SELECT s.pid,
     s.ssl_client_dn AS client_dn,
     s.ssl_client_serial AS client_serial,
     s.ssl_issuer_dn AS issuer_dn
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,

 doc/src/sgml/config.sgml                 |   2 +-
 doc/src/sgml/monitoring.sgml             |  14 ++++
 src/backend/access/heap/vacuumlazy.c     |   2 +-
 src/backend/access/nbtree/nbtsort.c      |   2 +-
 src/backend/catalog/system_views.sql     |   1 +
 src/backend/executor/execParallel.c      |  16 +++-
 src/backend/postmaster/autovacuum.c      |   4 +-
 src/backend/postmaster/pgstat.c          |   8 +-
 src/backend/replication/logical/worker.c |  14 ++--
 src/backend/replication/walsender.c      |   6 +-
 src/backend/tcop/postgres.c              |  31 ++++----
 src/backend/utils/adt/pgstatfuncs.c      | 132 ++++++++++++++++++-------------
 src/include/catalog/pg_proc.dat          |   6 +-
 src/include/pgstat.h                     |  15 +++-
 src/test/modules/worker_spi/worker_spi.c |   8 +-
 src/test/regress/expected/rules.out      |   9 ++-
 16 files changed, 173 insertions(+), 97 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index c4ba49ffaf..716aa65e15 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -7232,7 +7232,7 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv;
        <para>
        Specifies the amount of memory reserved to store the text of the
        currently executing command for each active session, for the
-       <structname>pg_stat_activity</structname>.<structfield>query</structfield> field.
+       <structname>pg_stat_activity</structname>.<structfield>query</structfield> and <structfield>individual_query</structfield> fields.
        If this value is specified without units, it is taken as bytes.
        The default value is 1024 bytes.
        This parameter can only be set at server start.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 673a0e73e4..b399e4891a 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -896,6 +896,20 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>individual_query</structfield> <type>text</type>
+      </para>
+      <para>
+       Text of this backend's most recent individual query in case <structfield>query</structfield> contains multiple statements. If
+       <structfield>state</structfield> is <literal>active</literal> this field shows the
+       currently executing individual query. In all other states, it shows the last individual query
+       that was executed. By default the individual query text is truncated at 1024
+       bytes; this value can be changed via the parameter
+       <xref linkend="guc-track-activity-query-size"/>.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>backend_type</structfield> <type>text</type>
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 92389e6666..45c72ba27c 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -3530,7 +3530,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	/* Set debug_query_string for individual workers */
 	sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, false);
 	debug_query_string = sharedquery;
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/*
 	 * Open table.  The lock mode is the same as the leader process.  It's
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index efee86784b..81725bd0ac 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1810,7 +1810,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
 	debug_query_string = sharedquery;
 
 	/* Report the query string from leader */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/* Look up nbtree shared state */
 	btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index ed4f3f142d..7eb7e4e4be 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -765,6 +765,7 @@ CREATE VIEW pg_stat_activity AS
             S.backend_xid,
             s.backend_xmin,
             S.query,
+            S.individual_query,
             S.backend_type
     FROM pg_stat_get_activity(NULL) AS S
         LEFT JOIN pg_database AS D ON (S.datid = D.oid)
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 382e78fb7f..e0587d57b4 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -65,6 +65,7 @@
 #define PARALLEL_KEY_QUERY_TEXT		UINT64CONST(0xE000000000000008)
 #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
 #define PARALLEL_KEY_WAL_USAGE			UINT64CONST(0xE00000000000000A)
+#define PARALLEL_KEY_INDIVIDUAL_QUERY	UINT64CONST(0xE00000000000000B)
 
 #define PARALLEL_TUPLE_QUEUE_SIZE		65536
 
@@ -601,6 +602,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	Size		dsa_minsize = dsa_minimum_size();
 	char	   *query_string;
 	int			query_len;
+	PgStat_IndividualQuery *individual_query;
 
 	/*
 	 * Force any initplan outputs that we're going to pass to workers to be
@@ -638,6 +640,10 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 						   sizeof(FixedParallelExecutorState));
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
+	/* Estimate space for individual query. */
+	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(PgStat_IndividualQuery));
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
 	/* Estimate space for query text. */
 	query_len = strlen(estate->es_sourceText);
 	shm_toc_estimate_chunk(&pcxt->estimator, query_len + 1);
@@ -732,6 +738,12 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	fpes->jit_flags = estate->es_jit_flags;
 	shm_toc_insert(pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fpes);
 
+	/* individual query stmt_len and stmt_location*/
+	individual_query = shm_toc_allocate(pcxt->toc, sizeof(PgStat_IndividualQuery));
+	individual_query->stmt_len = estate->es_plannedstmt->stmt_len;
+	individual_query->stmt_location = estate->es_plannedstmt->stmt_location;
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_INDIVIDUAL_QUERY, individual_query);
+
 	/* Store query string */
 	query_string = shm_toc_allocate(pcxt->toc, query_len + 1);
 	memcpy(query_string, estate->es_sourceText, query_len + 1);
@@ -1388,6 +1400,7 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 	void	   *area_space;
 	dsa_area   *area;
 	ParallelWorkerContext pwcxt;
+	PgStat_IndividualQuery *individual_query;
 
 	/* Get fixed-size state. */
 	fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
@@ -1403,9 +1416,10 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 
 	/* Setting debug_query_string for individual workers */
 	debug_query_string = queryDesc->sourceText;
+	individual_query = shm_toc_lookup(toc, PARALLEL_KEY_INDIVIDUAL_QUERY, false);
 
 	/* Report workers' query for monitoring purposes */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, individual_query->stmt_location, individual_query->stmt_len);
 
 	/* Attach to the dynamic shared memory area. */
 	area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 1b8cd7bacd..019da8b946 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -3184,7 +3184,7 @@ autovac_report_activity(autovac_table *tab)
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
@@ -3223,7 +3223,7 @@ autovac_report_workitem(AutoVacuumWorkItem *workitem,
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 5f4b168fd1..c5cea4efb5 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -3129,7 +3129,7 @@ pgstat_beshutdown_hook(int code, Datum arg)
  * ----------
  */
 void
-pgstat_report_activity(BackendState state, const char *cmd_str)
+pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len)
 {
 	volatile PgBackendStatus *beentry = MyBEEntry;
 	TimestampTz start_timestamp;
@@ -3156,6 +3156,8 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 			beentry->st_state = STATE_DISABLED;
 			beentry->st_state_start_timestamp = 0;
 			beentry->st_activity_raw[0] = '\0';
+			beentry->stmt_location = 0;
+			beentry->stmt_len = 0;
 			beentry->st_activity_start_timestamp = 0;
 			/* st_xact_start_timestamp and wait_event_info are also disabled */
 			beentry->st_xact_start_timestamp = 0;
@@ -3178,7 +3180,11 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 		 * than computed every write.
 		 */
 		len = Min(strlen(cmd_str), pgstat_track_activity_query_size - 1);
+		/* Individual query */
+		beentry->stmt_location = stmt_location;
+		beentry->stmt_len = stmt_len;
 	}
+
 	current_timestamp = GetCurrentTimestamp();
 
 	/*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index c37aafed0d..4520bed45b 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -702,7 +702,7 @@ apply_handle_begin(StringInfo s)
 
 	in_remote_transaction = true;
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 }
 
 /*
@@ -746,7 +746,7 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 /*
@@ -814,7 +814,7 @@ apply_handle_stream_start(StringInfo s)
 	if (!first_segment)
 		subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 }
 
 /*
@@ -843,7 +843,7 @@ apply_handle_stream_stop(StringInfo s)
 	/* Reset per-stream context */
 	MemoryContextReset(LogicalStreamingContext);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0 , 0);
 }
 
 /*
@@ -996,7 +996,7 @@ apply_handle_stream_commit(StringInfo s)
 	 * transaction.
 	 */
 	in_remote_transaction = true;
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0 ,0);
 
 	/*
 	 * Read the entries one by one and pass them through the same logic as in
@@ -1084,7 +1084,7 @@ apply_handle_stream_commit(StringInfo s)
 	/* unlink the files with serialized changes and subxact info */
 	stream_cleanup_files(MyLogicalRepWorker->subid, xid);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 /*
@@ -2079,7 +2079,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 													ALLOCSET_DEFAULT_SIZES);
 
 	/* mark as idle, before starting to loop */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 	/* This outer loop iterates once per wait. */
 	for (;;)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3f756b470a..2748d782e1 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1598,7 +1598,7 @@ exec_replication_command(const char *cmd_string)
 	initStringInfo(&tmpbuf);
 
 	/* Report to pgstat that this process is running */
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 
 	switch (cmd_node->type)
 	{
@@ -1657,7 +1657,7 @@ exec_replication_command(const char *cmd_string)
 						(errmsg("cannot execute SQL commands in WAL sender for physical replication")));
 
 			/* Report to pgstat that this process is now idle */
-			pgstat_report_activity(STATE_IDLE, NULL);
+			pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 			/* Tell the caller that this wasn't a WalSender command. */
 			return false;
@@ -1676,7 +1676,7 @@ exec_replication_command(const char *cmd_string)
 	EndCommand(&qc, DestRemote, true);
 
 	/* Report to pgstat that this process is now idle */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 	return true;
 }
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index c9424f167c..4cffc5540c 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -991,14 +991,6 @@ exec_simple_query(const char *query_string)
 	bool		use_implicit_block;
 	char		msec_str[32];
 
-	/*
-	 * Report query to various monitoring facilities.
-	 */
-	debug_query_string = query_string;
-
-	pgstat_report_activity(STATE_RUNNING, query_string);
-
-	TRACE_POSTGRESQL_QUERY_START(query_string);
 
 	/*
 	 * We use save_log_statement_stats so ShowUsage doesn't report incorrect
@@ -1076,6 +1068,15 @@ exec_simple_query(const char *query_string)
 		DestReceiver *receiver;
 		int16		format;
 
+		/*
+		 * Report query to various monitoring facilities.
+		 */
+		debug_query_string = query_string;
+
+		pgstat_report_activity(STATE_RUNNING, query_string, parsetree->stmt_location, parsetree->stmt_len);
+
+		TRACE_POSTGRESQL_QUERY_START(query_string);
+
 		/*
 		 * Get the command name for use in status display (it also becomes the
 		 * default completion tag, down inside PortalRun).  Set ps_status and
@@ -1366,7 +1367,7 @@ exec_parse_message(const char *query_string,	/* string to execute */
 	 */
 	debug_query_string = query_string;
 
-	pgstat_report_activity(STATE_RUNNING, query_string);
+	pgstat_report_activity(STATE_RUNNING, query_string, 0, 0);
 
 	set_ps_display("PARSE");
 
@@ -1657,7 +1658,7 @@ exec_bind_message(StringInfo input_message)
 	 */
 	debug_query_string = psrc->query_string;
 
-	pgstat_report_activity(STATE_RUNNING, psrc->query_string);
+	pgstat_report_activity(STATE_RUNNING, psrc->query_string, 0, 0);
 
 	set_ps_display("BIND");
 
@@ -2115,7 +2116,7 @@ exec_execute_message(const char *portal_name, long max_rows)
 	 */
 	debug_query_string = sourceText;
 
-	pgstat_report_activity(STATE_RUNNING, sourceText);
+	pgstat_report_activity(STATE_RUNNING, sourceText, 0, 0);
 
 	set_ps_display(GetCommandTagName(portal->commandTag));
 
@@ -4196,7 +4197,7 @@ PostgresMain(int argc, char *argv[],
 			if (IsAbortedTransactionBlockState())
 			{
 				set_ps_display("idle in transaction (aborted)");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4209,7 +4210,7 @@ PostgresMain(int argc, char *argv[],
 			else if (IsTransactionOrTransactionBlock())
 			{
 				set_ps_display("idle in transaction");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4236,7 +4237,7 @@ PostgresMain(int argc, char *argv[],
 				pgstat_report_stat(false);
 
 				set_ps_display("idle");
-				pgstat_report_activity(STATE_IDLE, NULL);
+				pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 			}
 
 			ReadyForQuery(whereToSendOutput);
@@ -4384,7 +4385,7 @@ PostgresMain(int argc, char *argv[],
 				SetCurrentStatementStartTimestamp();
 
 				/* Report query to various monitoring facilities. */
-				pgstat_report_activity(STATE_FASTPATH, NULL);
+				pgstat_report_activity(STATE_FASTPATH, NULL, 0, 0);
 				set_ps_display("<FASTPATH>");
 
 				/* start an xact for this function invocation */
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 95738a4e34..0700a227c6 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -567,7 +567,7 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_activity(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_ACTIVITY_COLS	30
+#define PG_STAT_GET_ACTIVITY_COLS	31
 	int			num_backends = pgstat_fetch_stat_numbackends();
 	int			curr_backend;
 	int			pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -631,6 +631,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			nulls[5] = false;
 			values[5] = CStringGetTextDatum("<backend information not available>");
+			nulls[6] = false;
+			values[6] = CStringGetTextDatum("<backend information not available>");
 
 			tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 			continue;
@@ -661,20 +663,21 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[3] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xid))
-			values[15] = TransactionIdGetDatum(local_beentry->backend_xid);
+			values[16] = TransactionIdGetDatum(local_beentry->backend_xid);
 		else
-			nulls[15] = true;
+			nulls[16] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xmin))
-			values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
+			values[17] = TransactionIdGetDatum(local_beentry->backend_xmin);
 		else
-			nulls[16] = true;
+			nulls[17] = true;
 
 		/* Values only available to role member or pg_read_all_stats */
 		if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
 		{
 			SockAddr	zero_clientaddr;
 			char	   *clipped_activity;
+			int		individual_len;
 
 			switch (beentry->st_state)
 			{
@@ -703,10 +706,32 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			clipped_activity = pgstat_clip_activity(beentry->st_activity_raw);
 			values[5] = CStringGetTextDatum(clipped_activity);
+
+			if (clipped_activity[0] != '\0')
+			{
+			/* display individual query */
+				if ((beentry->stmt_len + beentry->stmt_location < pgstat_track_activity_query_size)
+					&& (beentry->stmt_len + 1 != strlen(clipped_activity))
+					&& (beentry->stmt_len + beentry->stmt_location != 0))
+				{
+					if (beentry->stmt_len ==  0)
+						individual_len = strlen(clipped_activity) - beentry->stmt_location;
+					else
+						individual_len = beentry->stmt_len + 1;
+
+					clipped_activity[beentry->stmt_location + individual_len] = '\0';
+					values[6] = CStringGetTextDatum(clipped_activity + beentry->stmt_location);
+				} else {
+					nulls[6] = true;
+				}
+			} else {
+				nulls[6] = true;
+			}
+
 			pfree(clipped_activity);
 
 			/* leader_pid */
-			nulls[29] = true;
+			nulls[30] = true;
 
 			proc = BackendPidGetProc(beentry->st_procpid);
 
@@ -743,20 +768,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 				 */
 				if (leader && leader->pid != beentry->st_procpid)
 				{
-					values[29] = Int32GetDatum(leader->pid);
-					nulls[29] = false;
+					values[30] = Int32GetDatum(leader->pid);
+					nulls[30] = false;
 				}
 			}
 
 			if (wait_event_type)
-				values[6] = CStringGetTextDatum(wait_event_type);
+				values[7] = CStringGetTextDatum(wait_event_type);
 			else
-				nulls[6] = true;
+				nulls[7] = true;
 
 			if (wait_event)
-				values[7] = CStringGetTextDatum(wait_event);
+				values[8] = CStringGetTextDatum(wait_event);
 			else
-				nulls[7] = true;
+				nulls[8] = true;
 
 			/*
 			 * Don't expose transaction time for walsenders; it confuses
@@ -765,33 +790,33 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			 */
 			if (beentry->st_xact_start_timestamp != 0 &&
 				beentry->st_backendType != B_WAL_SENDER)
-				values[8] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
+				values[9] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
 			else
-				nulls[8] = true;
+				nulls[9] = true;
 
 			if (beentry->st_activity_start_timestamp != 0)
-				values[9] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
+				values[10] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
 			else
-				nulls[9] = true;
+				nulls[10] = true;
 
 			if (beentry->st_proc_start_timestamp != 0)
-				values[10] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
+				values[11] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
 			else
-				nulls[10] = true;
+				nulls[11] = true;
 
 			if (beentry->st_state_start_timestamp != 0)
-				values[11] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
+				values[12] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
 			else
-				nulls[11] = true;
+				nulls[12] = true;
 
 			/* A zeroed client addr means we don't know */
 			memset(&zero_clientaddr, 0, sizeof(zero_clientaddr));
 			if (memcmp(&(beentry->st_clientaddr), &zero_clientaddr,
 					   sizeof(zero_clientaddr)) == 0)
 			{
-				nulls[12] = true;
 				nulls[13] = true;
 				nulls[14] = true;
+				nulls[15] = true;
 			}
 			else
 			{
@@ -815,20 +840,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					if (ret == 0)
 					{
 						clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);
-						values[12] = DirectFunctionCall1(inet_in,
+						values[13] = DirectFunctionCall1(inet_in,
 														 CStringGetDatum(remote_host));
 						if (beentry->st_clienthostname &&
 							beentry->st_clienthostname[0])
-							values[13] = CStringGetTextDatum(beentry->st_clienthostname);
+							values[14] = CStringGetTextDatum(beentry->st_clienthostname);
 						else
-							nulls[13] = true;
-						values[14] = Int32GetDatum(atoi(remote_port));
+							nulls[14] = true;
+						values[15] = Int32GetDatum(atoi(remote_port));
 					}
 					else
 					{
-						nulls[12] = true;
 						nulls[13] = true;
 						nulls[14] = true;
+						nulls[15] = true;
 					}
 				}
 				else if (beentry->st_clientaddr.addr.ss_family == AF_UNIX)
@@ -839,16 +864,16 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					 * connections we have no permissions to view, or with
 					 * errors.
 					 */
-					nulls[12] = true;
 					nulls[13] = true;
-					values[14] = Int32GetDatum(-1);
+					nulls[14] = true;
+					values[15] = Int32GetDatum(-1);
 				}
 				else
 				{
 					/* Unknown address type, should never happen */
-					nulls[12] = true;
 					nulls[13] = true;
 					nulls[14] = true;
+					nulls[15] = true;
 				}
 			}
 			/* Add backend type */
@@ -858,59 +883,59 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 				bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
 				if (bgw_type)
-					values[17] = CStringGetTextDatum(bgw_type);
+					values[18] = CStringGetTextDatum(bgw_type);
 				else
-					nulls[17] = true;
+					nulls[18] = true;
 			}
 			else
-				values[17] =
+				values[18] =
 					CStringGetTextDatum(GetBackendTypeDesc(beentry->st_backendType));
 
 			/* SSL information */
 			if (beentry->st_ssl)
 			{
-				values[18] = BoolGetDatum(true);	/* ssl */
-				values[19] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
-				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
-				values[21] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
-				values[22] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
+				values[19] = BoolGetDatum(true);	/* ssl */
+				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
+				values[21] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
+				values[22] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
+				values[23] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
 
 				if (beentry->st_sslstatus->ssl_client_dn[0])
-					values[23] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
+					values[24] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
 				else
-					nulls[23] = true;
+					nulls[24] = true;
 
 				if (beentry->st_sslstatus->ssl_client_serial[0])
-					values[24] = DirectFunctionCall3(numeric_in,
+					values[25] = DirectFunctionCall3(numeric_in,
 													 CStringGetDatum(beentry->st_sslstatus->ssl_client_serial),
 													 ObjectIdGetDatum(InvalidOid),
 													 Int32GetDatum(-1));
 				else
-					nulls[24] = true;
+					nulls[25] = true;
 
 				if (beentry->st_sslstatus->ssl_issuer_dn[0])
-					values[25] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
+					values[26] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
 				else
-					nulls[25] = true;
+					nulls[26] = true;
 			}
 			else
 			{
-				values[18] = BoolGetDatum(false);	/* ssl */
-				nulls[19] = nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = true;
+				values[19] = BoolGetDatum(false);	/* ssl */
+				nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = nulls[26] = true;
 			}
 
 			/* GSSAPI information */
 			if (beentry->st_gss)
 			{
-				values[26] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
-				values[27] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
-				values[28] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
+				values[27] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
+				values[28] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
+				values[29] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
 			}
 			else
 			{
-				values[26] = BoolGetDatum(false);	/* gss_auth */
-				nulls[27] = true;	/* No GSS principal */
-				values[28] = BoolGetDatum(false);	/* GSS Encryption not in
+				values[27] = BoolGetDatum(false);	/* gss_auth */
+				nulls[28] = true;	/* No GSS principal */
+				values[29] = BoolGetDatum(false);	/* GSS Encryption not in
 													 * use */
 			}
 		}
@@ -918,8 +943,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 		{
 			/* No permissions to view data about this session */
 			values[5] = CStringGetTextDatum("<insufficient privilege>");
+			values[6] = CStringGetTextDatum("<insufficient privilege>");
 			nulls[4] = true;
-			nulls[6] = true;
 			nulls[7] = true;
 			nulls[8] = true;
 			nulls[9] = true;
@@ -928,7 +953,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[12] = true;
 			nulls[13] = true;
 			nulls[14] = true;
-			nulls[17] = true;
+			nulls[15] = true;
 			nulls[18] = true;
 			nulls[19] = true;
 			nulls[20] = true;
@@ -941,6 +966,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[27] = true;
 			nulls[28] = true;
 			nulls[29] = true;
+			nulls[30] = true;
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 687509ba92..7cd085faf9 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5231,9 +5231,9 @@
   proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'int4',
-  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
+  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,individual_query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
   prosrc => 'pg_stat_get_activity' },
 { oid => '3318',
   descr => 'statistics: information about progress of backends running maintenance command',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 0dfbac46b4..a303c66cdf 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -118,6 +118,13 @@ typedef struct PgStat_TableCounts
 	PgStat_Counter t_blocks_hit;
 } PgStat_TableCounts;
 
+/* Individual Query location and length*/
+typedef struct PgStat_IndividualQuery
+{
+	int         stmt_location;
+	int         stmt_len;
+} PgStat_IndividualQuery;
+
 /* Possible targets for resetting cluster-wide shared values */
 typedef enum PgStat_Shared_Reset_Target
 {
@@ -1128,6 +1135,12 @@ typedef struct PgBackendStatus
 	 */
 	char	   *st_activity_raw;
 
+	/*
+	 * Current individual command location and length
+	 */
+	int stmt_location;
+	int stmt_len;
+
 	/*
 	 * Command progress reporting.  Any command which wishes can advertise
 	 * that it is running by setting st_progress_command,
@@ -1319,7 +1332,7 @@ extern void pgstat_report_checksum_failure(void);
 extern void pgstat_initialize(void);
 extern void pgstat_bestart(void);
 
-extern void pgstat_report_activity(BackendState state, const char *cmd_str);
+extern void pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len);
 extern void pgstat_report_tempfile(size_t filesize);
 extern void pgstat_report_appname(const char *appname);
 extern void pgstat_report_xact_timestamp(TimestampTz tstamp);
diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index 1c7b17c56f..c4d528f291 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -112,7 +112,7 @@ initialize_worker_spi(worktable *table)
 	StartTransactionCommand();
 	SPI_connect();
 	PushActiveSnapshot(GetTransactionSnapshot());
-	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
+	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema", 0, 0);
 
 	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
 	initStringInfo(&buf);
@@ -156,7 +156,7 @@ initialize_worker_spi(worktable *table)
 	SPI_finish();
 	PopActiveSnapshot();
 	CommitTransactionCommand();
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 void
@@ -262,7 +262,7 @@ worker_spi_main(Datum main_arg)
 		StartTransactionCommand();
 		SPI_connect();
 		PushActiveSnapshot(GetTransactionSnapshot());
-		pgstat_report_activity(STATE_RUNNING, buf.data);
+		pgstat_report_activity(STATE_RUNNING, buf.data, 0, 0);
 
 		/* We can now execute queries via SPI */
 		ret = SPI_execute(buf.data, false, 0);
@@ -292,7 +292,7 @@ worker_spi_main(Datum main_arg)
 		PopActiveSnapshot();
 		CommitTransactionCommand();
 		pgstat_report_stat(false);
-		pgstat_report_activity(STATE_IDLE, NULL);
+		pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 	}
 
 	proc_exit(1);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2a18dc423e..313993c392 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1761,8 +1761,9 @@ pg_stat_activity| SELECT s.datid,
     s.backend_xid,
     s.backend_xmin,
     s.query,
+    s.individual_query,
     s.backend_type
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1867,7 +1868,7 @@ pg_stat_gssapi| SELECT s.pid,
     s.gss_auth AS gss_authenticated,
     s.gss_princ AS principal,
     s.gss_enc AS encrypted
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_progress_analyze| SELECT s.pid,
     s.datid,
@@ -2015,7 +2016,7 @@ pg_stat_replication| SELECT s.pid,
     w.sync_priority,
     w.sync_state,
     w.reply_time
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_slru| SELECT s.name,
@@ -2037,7 +2038,7 @@ pg_stat_ssl| SELECT s.pid,
     s.ssl_client_dn AS client_dn,
     s.ssl_client_serial AS client_serial,
     s.ssl_issuer_dn AS issuer_dn
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,

 doc/src/sgml/config.sgml                 |   2 +-
 doc/src/sgml/monitoring.sgml             |  14 ++++
 src/backend/access/heap/vacuumlazy.c     |   2 +-
 src/backend/access/nbtree/nbtsort.c      |   2 +-
 src/backend/catalog/system_views.sql     |   1 +
 src/backend/executor/execParallel.c      |  16 +++-
 src/backend/postmaster/autovacuum.c      |   4 +-
 src/backend/postmaster/pgstat.c          |   8 +-
 src/backend/replication/logical/worker.c |  14 ++--
 src/backend/replication/walsender.c      |   2 +-
 src/backend/tcop/postgres.c              |  31 ++++----
 src/backend/utils/adt/pgstatfuncs.c      | 132 ++++++++++++++++++-------------
 src/include/catalog/pg_proc.dat          |   6 +-
 src/include/pgstat.h                     |  15 +++-
 src/test/modules/worker_spi/worker_spi.c |   8 +-
 src/test/regress/expected/rules.out      |   9 ++-
 16 files changed, 171 insertions(+), 95 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 8eabf93834..ad70e26049 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -7234,7 +7234,7 @@ COPY postgres_log FROM '/full/path/to/logfile.csv' WITH csv;
        <para>
        Specifies the amount of memory reserved to store the text of the
        currently executing command for each active session, for the
-       <structname>pg_stat_activity</structname>.<structfield>query</structfield> field.
+       <structname>pg_stat_activity</structname>.<structfield>query</structfield> and <structfield>individual_query</structfield> fields.
        If this value is specified without units, it is taken as bytes.
        The default value is 1024 bytes.
        This parameter can only be set at server start.
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 4e0193a967..a15e4b1367 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -896,6 +896,20 @@ postgres   27093  0.0  0.0  30096  2752 ?        Ss   11:34   0:00 postgres: ser
       </para></entry>
      </row>
 
+     <row>
+      <entry role="catalog_table_entry"><para role="column_definition">
+       <structfield>individual_query</structfield> <type>text</type>
+      </para>
+      <para>
+       Text of this backend's most recent individual query in case <structfield>query</structfield> contains multiple statements. If
+       <structfield>state</structfield> is <literal>active</literal> this field shows the
+       currently executing individual query. In all other states, it shows the last individual query
+       that was executed. By default the individual query text is truncated at 1024
+       bytes; this value can be changed via the parameter
+       <xref linkend="guc-track-activity-query-size"/>.
+      </para></entry>
+     </row>
+
      <row>
       <entry role="catalog_table_entry"><para role="column_definition">
        <structfield>backend_type</structfield> <type>text</type>
diff --git a/src/backend/access/heap/vacuumlazy.c b/src/backend/access/heap/vacuumlazy.c
index 4f2f38168d..0d53ce0cf9 100644
--- a/src/backend/access/heap/vacuumlazy.c
+++ b/src/backend/access/heap/vacuumlazy.c
@@ -3530,7 +3530,7 @@ parallel_vacuum_main(dsm_segment *seg, shm_toc *toc)
 	/* Set debug_query_string for individual workers */
 	sharedquery = shm_toc_lookup(toc, PARALLEL_VACUUM_KEY_QUERY_TEXT, false);
 	debug_query_string = sharedquery;
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/*
 	 * Open table.  The lock mode is the same as the leader process.  It's
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
index efee86784b..81725bd0ac 100644
--- a/src/backend/access/nbtree/nbtsort.c
+++ b/src/backend/access/nbtree/nbtsort.c
@@ -1810,7 +1810,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
 	debug_query_string = sharedquery;
 
 	/* Report the query string from leader */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, 0, 0);
 
 	/* Look up nbtree shared state */
 	btshared = shm_toc_lookup(toc, PARALLEL_KEY_BTREE_SHARED, false);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index ed4f3f142d..7eb7e4e4be 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -765,6 +765,7 @@ CREATE VIEW pg_stat_activity AS
             S.backend_xid,
             s.backend_xmin,
             S.query,
+            S.individual_query,
             S.backend_type
     FROM pg_stat_get_activity(NULL) AS S
         LEFT JOIN pg_database AS D ON (S.datid = D.oid)
diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c
index 382e78fb7f..e0587d57b4 100644
--- a/src/backend/executor/execParallel.c
+++ b/src/backend/executor/execParallel.c
@@ -65,6 +65,7 @@
 #define PARALLEL_KEY_QUERY_TEXT		UINT64CONST(0xE000000000000008)
 #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
 #define PARALLEL_KEY_WAL_USAGE			UINT64CONST(0xE00000000000000A)
+#define PARALLEL_KEY_INDIVIDUAL_QUERY	UINT64CONST(0xE00000000000000B)
 
 #define PARALLEL_TUPLE_QUEUE_SIZE		65536
 
@@ -601,6 +602,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	Size		dsa_minsize = dsa_minimum_size();
 	char	   *query_string;
 	int			query_len;
+	PgStat_IndividualQuery *individual_query;
 
 	/*
 	 * Force any initplan outputs that we're going to pass to workers to be
@@ -638,6 +640,10 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 						   sizeof(FixedParallelExecutorState));
 	shm_toc_estimate_keys(&pcxt->estimator, 1);
 
+	/* Estimate space for individual query. */
+	shm_toc_estimate_chunk(&pcxt->estimator, sizeof(PgStat_IndividualQuery));
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
 	/* Estimate space for query text. */
 	query_len = strlen(estate->es_sourceText);
 	shm_toc_estimate_chunk(&pcxt->estimator, query_len + 1);
@@ -732,6 +738,12 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate,
 	fpes->jit_flags = estate->es_jit_flags;
 	shm_toc_insert(pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fpes);
 
+	/* individual query stmt_len and stmt_location*/
+	individual_query = shm_toc_allocate(pcxt->toc, sizeof(PgStat_IndividualQuery));
+	individual_query->stmt_len = estate->es_plannedstmt->stmt_len;
+	individual_query->stmt_location = estate->es_plannedstmt->stmt_location;
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_INDIVIDUAL_QUERY, individual_query);
+
 	/* Store query string */
 	query_string = shm_toc_allocate(pcxt->toc, query_len + 1);
 	memcpy(query_string, estate->es_sourceText, query_len + 1);
@@ -1388,6 +1400,7 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 	void	   *area_space;
 	dsa_area   *area;
 	ParallelWorkerContext pwcxt;
+	PgStat_IndividualQuery *individual_query;
 
 	/* Get fixed-size state. */
 	fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false);
@@ -1403,9 +1416,10 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc)
 
 	/* Setting debug_query_string for individual workers */
 	debug_query_string = queryDesc->sourceText;
+	individual_query = shm_toc_lookup(toc, PARALLEL_KEY_INDIVIDUAL_QUERY, false);
 
 	/* Report workers' query for monitoring purposes */
-	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+	pgstat_report_activity(STATE_RUNNING, debug_query_string, individual_query->stmt_location, individual_query->stmt_len);
 
 	/* Attach to the dynamic shared memory area. */
 	area_space = shm_toc_lookup(toc, PARALLEL_KEY_DSA, false);
diff --git a/src/backend/postmaster/autovacuum.c b/src/backend/postmaster/autovacuum.c
index 2cef56f115..3433ff9826 100644
--- a/src/backend/postmaster/autovacuum.c
+++ b/src/backend/postmaster/autovacuum.c
@@ -3200,7 +3200,7 @@ autovac_report_activity(autovac_table *tab)
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
@@ -3239,7 +3239,7 @@ autovac_report_workitem(AutoVacuumWorkItem *workitem,
 	/* Set statement_timestamp() to current time for pg_stat_activity */
 	SetCurrentStatementStartTimestamp();
 
-	pgstat_report_activity(STATE_RUNNING, activity);
+	pgstat_report_activity(STATE_RUNNING, activity, 0, 0);
 }
 
 /*
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index e6be2b7836..74ad2646f0 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -3129,7 +3129,7 @@ pgstat_beshutdown_hook(int code, Datum arg)
  * ----------
  */
 void
-pgstat_report_activity(BackendState state, const char *cmd_str)
+pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len)
 {
 	volatile PgBackendStatus *beentry = MyBEEntry;
 	TimestampTz start_timestamp;
@@ -3156,6 +3156,8 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 			beentry->st_state = STATE_DISABLED;
 			beentry->st_state_start_timestamp = 0;
 			beentry->st_activity_raw[0] = '\0';
+			beentry->stmt_location = 0;
+			beentry->stmt_len = 0;
 			beentry->st_activity_start_timestamp = 0;
 			/* st_xact_start_timestamp and wait_event_info are also disabled */
 			beentry->st_xact_start_timestamp = 0;
@@ -3178,7 +3180,11 @@ pgstat_report_activity(BackendState state, const char *cmd_str)
 		 * than computed every write.
 		 */
 		len = Min(strlen(cmd_str), pgstat_track_activity_query_size - 1);
+		/* Individual query */
+		beentry->stmt_location = stmt_location;
+		beentry->stmt_len = stmt_len;
 	}
+
 	current_timestamp = GetCurrentTimestamp();
 
 	/*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 9c6fdeeb56..1f179fc13d 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -702,7 +702,7 @@ apply_handle_begin(StringInfo s)
 
 	in_remote_transaction = true;
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 }
 
 /*
@@ -746,7 +746,7 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 /*
@@ -814,7 +814,7 @@ apply_handle_stream_start(StringInfo s)
 	if (!first_segment)
 		subxact_info_read(MyLogicalRepWorker->subid, stream_xid);
 
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0, 0);
 }
 
 /*
@@ -843,7 +843,7 @@ apply_handle_stream_stop(StringInfo s)
 	/* Reset per-stream context */
 	MemoryContextReset(LogicalStreamingContext);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0 , 0);
 }
 
 /*
@@ -996,7 +996,7 @@ apply_handle_stream_commit(StringInfo s)
 	 * transaction.
 	 */
 	in_remote_transaction = true;
-	pgstat_report_activity(STATE_RUNNING, NULL);
+	pgstat_report_activity(STATE_RUNNING, NULL, 0 ,0);
 
 	/*
 	 * Read the entries one by one and pass them through the same logic as in
@@ -1084,7 +1084,7 @@ apply_handle_stream_commit(StringInfo s)
 	/* unlink the files with serialized changes and subxact info */
 	stream_cleanup_files(MyLogicalRepWorker->subid, xid);
 
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 /*
@@ -2079,7 +2079,7 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 													ALLOCSET_DEFAULT_SIZES);
 
 	/* mark as idle, before starting to loop */
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 
 	/* This outer loop iterates once per wait. */
 	for (;;)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 7c9d1b67df..116696399a 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1583,7 +1583,7 @@ exec_replication_command(const char *cmd_string)
 	 */
 	debug_query_string = cmd_string;
 
-	pgstat_report_activity(STATE_RUNNING, cmd_string);
+	pgstat_report_activity(STATE_RUNNING, cmd_string, 0, 0);
 
 	/*
 	 * Log replication command if log_replication_commands is enabled. Even
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 411cfadbff..36c76375ca 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -991,14 +991,6 @@ exec_simple_query(const char *query_string)
 	bool		use_implicit_block;
 	char		msec_str[32];
 
-	/*
-	 * Report query to various monitoring facilities.
-	 */
-	debug_query_string = query_string;
-
-	pgstat_report_activity(STATE_RUNNING, query_string);
-
-	TRACE_POSTGRESQL_QUERY_START(query_string);
 
 	/*
 	 * We use save_log_statement_stats so ShowUsage doesn't report incorrect
@@ -1076,6 +1068,15 @@ exec_simple_query(const char *query_string)
 		DestReceiver *receiver;
 		int16		format;
 
+		/*
+		 * Report query to various monitoring facilities.
+		 */
+		debug_query_string = query_string;
+
+		pgstat_report_activity(STATE_RUNNING, query_string, parsetree->stmt_location, parsetree->stmt_len);
+
+		TRACE_POSTGRESQL_QUERY_START(query_string);
+
 		/*
 		 * Get the command name for use in status display (it also becomes the
 		 * default completion tag, down inside PortalRun).  Set ps_status and
@@ -1366,7 +1367,7 @@ exec_parse_message(const char *query_string,	/* string to execute */
 	 */
 	debug_query_string = query_string;
 
-	pgstat_report_activity(STATE_RUNNING, query_string);
+	pgstat_report_activity(STATE_RUNNING, query_string, 0, 0);
 
 	set_ps_display("PARSE");
 
@@ -1657,7 +1658,7 @@ exec_bind_message(StringInfo input_message)
 	 */
 	debug_query_string = psrc->query_string;
 
-	pgstat_report_activity(STATE_RUNNING, psrc->query_string);
+	pgstat_report_activity(STATE_RUNNING, psrc->query_string, 0, 0);
 
 	set_ps_display("BIND");
 
@@ -2115,7 +2116,7 @@ exec_execute_message(const char *portal_name, long max_rows)
 	 */
 	debug_query_string = sourceText;
 
-	pgstat_report_activity(STATE_RUNNING, sourceText);
+	pgstat_report_activity(STATE_RUNNING, sourceText, 0, 0);
 
 	set_ps_display(GetCommandTagName(portal->commandTag));
 
@@ -4190,7 +4191,7 @@ PostgresMain(int argc, char *argv[],
 			if (IsAbortedTransactionBlockState())
 			{
 				set_ps_display("idle in transaction (aborted)");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4203,7 +4204,7 @@ PostgresMain(int argc, char *argv[],
 			else if (IsTransactionOrTransactionBlock())
 			{
 				set_ps_display("idle in transaction");
-				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
+				pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL, 0, 0);
 
 				/* Start the idle-in-transaction timer */
 				if (IdleInTransactionSessionTimeout > 0)
@@ -4230,7 +4231,7 @@ PostgresMain(int argc, char *argv[],
 				pgstat_report_stat(false);
 
 				set_ps_display("idle");
-				pgstat_report_activity(STATE_IDLE, NULL);
+				pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 			}
 
 			ReadyForQuery(whereToSendOutput);
@@ -4378,7 +4379,7 @@ PostgresMain(int argc, char *argv[],
 				SetCurrentStatementStartTimestamp();
 
 				/* Report query to various monitoring facilities. */
-				pgstat_report_activity(STATE_FASTPATH, NULL);
+				pgstat_report_activity(STATE_FASTPATH, NULL, 0, 0);
 				set_ps_display("<FASTPATH>");
 
 				/* start an xact for this function invocation */
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index 95738a4e34..0700a227c6 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -567,7 +567,7 @@ pg_stat_get_progress_info(PG_FUNCTION_ARGS)
 Datum
 pg_stat_get_activity(PG_FUNCTION_ARGS)
 {
-#define PG_STAT_GET_ACTIVITY_COLS	30
+#define PG_STAT_GET_ACTIVITY_COLS	31
 	int			num_backends = pgstat_fetch_stat_numbackends();
 	int			curr_backend;
 	int			pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -631,6 +631,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			nulls[5] = false;
 			values[5] = CStringGetTextDatum("<backend information not available>");
+			nulls[6] = false;
+			values[6] = CStringGetTextDatum("<backend information not available>");
 
 			tuplestore_putvalues(tupstore, tupdesc, values, nulls);
 			continue;
@@ -661,20 +663,21 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[3] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xid))
-			values[15] = TransactionIdGetDatum(local_beentry->backend_xid);
+			values[16] = TransactionIdGetDatum(local_beentry->backend_xid);
 		else
-			nulls[15] = true;
+			nulls[16] = true;
 
 		if (TransactionIdIsValid(local_beentry->backend_xmin))
-			values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
+			values[17] = TransactionIdGetDatum(local_beentry->backend_xmin);
 		else
-			nulls[16] = true;
+			nulls[17] = true;
 
 		/* Values only available to role member or pg_read_all_stats */
 		if (HAS_PGSTAT_PERMISSIONS(beentry->st_userid))
 		{
 			SockAddr	zero_clientaddr;
 			char	   *clipped_activity;
+			int		individual_len;
 
 			switch (beentry->st_state)
 			{
@@ -703,10 +706,32 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 			clipped_activity = pgstat_clip_activity(beentry->st_activity_raw);
 			values[5] = CStringGetTextDatum(clipped_activity);
+
+			if (clipped_activity[0] != '\0')
+			{
+			/* display individual query */
+				if ((beentry->stmt_len + beentry->stmt_location < pgstat_track_activity_query_size)
+					&& (beentry->stmt_len + 1 != strlen(clipped_activity))
+					&& (beentry->stmt_len + beentry->stmt_location != 0))
+				{
+					if (beentry->stmt_len ==  0)
+						individual_len = strlen(clipped_activity) - beentry->stmt_location;
+					else
+						individual_len = beentry->stmt_len + 1;
+
+					clipped_activity[beentry->stmt_location + individual_len] = '\0';
+					values[6] = CStringGetTextDatum(clipped_activity + beentry->stmt_location);
+				} else {
+					nulls[6] = true;
+				}
+			} else {
+				nulls[6] = true;
+			}
+
 			pfree(clipped_activity);
 
 			/* leader_pid */
-			nulls[29] = true;
+			nulls[30] = true;
 
 			proc = BackendPidGetProc(beentry->st_procpid);
 
@@ -743,20 +768,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 				 */
 				if (leader && leader->pid != beentry->st_procpid)
 				{
-					values[29] = Int32GetDatum(leader->pid);
-					nulls[29] = false;
+					values[30] = Int32GetDatum(leader->pid);
+					nulls[30] = false;
 				}
 			}
 
 			if (wait_event_type)
-				values[6] = CStringGetTextDatum(wait_event_type);
+				values[7] = CStringGetTextDatum(wait_event_type);
 			else
-				nulls[6] = true;
+				nulls[7] = true;
 
 			if (wait_event)
-				values[7] = CStringGetTextDatum(wait_event);
+				values[8] = CStringGetTextDatum(wait_event);
 			else
-				nulls[7] = true;
+				nulls[8] = true;
 
 			/*
 			 * Don't expose transaction time for walsenders; it confuses
@@ -765,33 +790,33 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			 */
 			if (beentry->st_xact_start_timestamp != 0 &&
 				beentry->st_backendType != B_WAL_SENDER)
-				values[8] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
+				values[9] = TimestampTzGetDatum(beentry->st_xact_start_timestamp);
 			else
-				nulls[8] = true;
+				nulls[9] = true;
 
 			if (beentry->st_activity_start_timestamp != 0)
-				values[9] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
+				values[10] = TimestampTzGetDatum(beentry->st_activity_start_timestamp);
 			else
-				nulls[9] = true;
+				nulls[10] = true;
 
 			if (beentry->st_proc_start_timestamp != 0)
-				values[10] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
+				values[11] = TimestampTzGetDatum(beentry->st_proc_start_timestamp);
 			else
-				nulls[10] = true;
+				nulls[11] = true;
 
 			if (beentry->st_state_start_timestamp != 0)
-				values[11] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
+				values[12] = TimestampTzGetDatum(beentry->st_state_start_timestamp);
 			else
-				nulls[11] = true;
+				nulls[12] = true;
 
 			/* A zeroed client addr means we don't know */
 			memset(&zero_clientaddr, 0, sizeof(zero_clientaddr));
 			if (memcmp(&(beentry->st_clientaddr), &zero_clientaddr,
 					   sizeof(zero_clientaddr)) == 0)
 			{
-				nulls[12] = true;
 				nulls[13] = true;
 				nulls[14] = true;
+				nulls[15] = true;
 			}
 			else
 			{
@@ -815,20 +840,20 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					if (ret == 0)
 					{
 						clean_ipv6_addr(beentry->st_clientaddr.addr.ss_family, remote_host);
-						values[12] = DirectFunctionCall1(inet_in,
+						values[13] = DirectFunctionCall1(inet_in,
 														 CStringGetDatum(remote_host));
 						if (beentry->st_clienthostname &&
 							beentry->st_clienthostname[0])
-							values[13] = CStringGetTextDatum(beentry->st_clienthostname);
+							values[14] = CStringGetTextDatum(beentry->st_clienthostname);
 						else
-							nulls[13] = true;
-						values[14] = Int32GetDatum(atoi(remote_port));
+							nulls[14] = true;
+						values[15] = Int32GetDatum(atoi(remote_port));
 					}
 					else
 					{
-						nulls[12] = true;
 						nulls[13] = true;
 						nulls[14] = true;
+						nulls[15] = true;
 					}
 				}
 				else if (beentry->st_clientaddr.addr.ss_family == AF_UNIX)
@@ -839,16 +864,16 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 					 * connections we have no permissions to view, or with
 					 * errors.
 					 */
-					nulls[12] = true;
 					nulls[13] = true;
-					values[14] = Int32GetDatum(-1);
+					nulls[14] = true;
+					values[15] = Int32GetDatum(-1);
 				}
 				else
 				{
 					/* Unknown address type, should never happen */
-					nulls[12] = true;
 					nulls[13] = true;
 					nulls[14] = true;
+					nulls[15] = true;
 				}
 			}
 			/* Add backend type */
@@ -858,59 +883,59 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 
 				bgw_type = GetBackgroundWorkerTypeByPid(beentry->st_procpid);
 				if (bgw_type)
-					values[17] = CStringGetTextDatum(bgw_type);
+					values[18] = CStringGetTextDatum(bgw_type);
 				else
-					nulls[17] = true;
+					nulls[18] = true;
 			}
 			else
-				values[17] =
+				values[18] =
 					CStringGetTextDatum(GetBackendTypeDesc(beentry->st_backendType));
 
 			/* SSL information */
 			if (beentry->st_ssl)
 			{
-				values[18] = BoolGetDatum(true);	/* ssl */
-				values[19] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
-				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
-				values[21] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
-				values[22] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
+				values[19] = BoolGetDatum(true);	/* ssl */
+				values[20] = CStringGetTextDatum(beentry->st_sslstatus->ssl_version);
+				values[21] = CStringGetTextDatum(beentry->st_sslstatus->ssl_cipher);
+				values[22] = Int32GetDatum(beentry->st_sslstatus->ssl_bits);
+				values[23] = BoolGetDatum(beentry->st_sslstatus->ssl_compression);
 
 				if (beentry->st_sslstatus->ssl_client_dn[0])
-					values[23] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
+					values[24] = CStringGetTextDatum(beentry->st_sslstatus->ssl_client_dn);
 				else
-					nulls[23] = true;
+					nulls[24] = true;
 
 				if (beentry->st_sslstatus->ssl_client_serial[0])
-					values[24] = DirectFunctionCall3(numeric_in,
+					values[25] = DirectFunctionCall3(numeric_in,
 													 CStringGetDatum(beentry->st_sslstatus->ssl_client_serial),
 													 ObjectIdGetDatum(InvalidOid),
 													 Int32GetDatum(-1));
 				else
-					nulls[24] = true;
+					nulls[25] = true;
 
 				if (beentry->st_sslstatus->ssl_issuer_dn[0])
-					values[25] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
+					values[26] = CStringGetTextDatum(beentry->st_sslstatus->ssl_issuer_dn);
 				else
-					nulls[25] = true;
+					nulls[26] = true;
 			}
 			else
 			{
-				values[18] = BoolGetDatum(false);	/* ssl */
-				nulls[19] = nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = true;
+				values[19] = BoolGetDatum(false);	/* ssl */
+				nulls[20] = nulls[21] = nulls[22] = nulls[23] = nulls[24] = nulls[25] = nulls[26] = true;
 			}
 
 			/* GSSAPI information */
 			if (beentry->st_gss)
 			{
-				values[26] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
-				values[27] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
-				values[28] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
+				values[27] = BoolGetDatum(beentry->st_gssstatus->gss_auth); /* gss_auth */
+				values[28] = CStringGetTextDatum(beentry->st_gssstatus->gss_princ);
+				values[29] = BoolGetDatum(beentry->st_gssstatus->gss_enc);	/* GSS Encryption in use */
 			}
 			else
 			{
-				values[26] = BoolGetDatum(false);	/* gss_auth */
-				nulls[27] = true;	/* No GSS principal */
-				values[28] = BoolGetDatum(false);	/* GSS Encryption not in
+				values[27] = BoolGetDatum(false);	/* gss_auth */
+				nulls[28] = true;	/* No GSS principal */
+				values[29] = BoolGetDatum(false);	/* GSS Encryption not in
 													 * use */
 			}
 		}
@@ -918,8 +943,8 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 		{
 			/* No permissions to view data about this session */
 			values[5] = CStringGetTextDatum("<insufficient privilege>");
+			values[6] = CStringGetTextDatum("<insufficient privilege>");
 			nulls[4] = true;
-			nulls[6] = true;
 			nulls[7] = true;
 			nulls[8] = true;
 			nulls[9] = true;
@@ -928,7 +953,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[12] = true;
 			nulls[13] = true;
 			nulls[14] = true;
-			nulls[17] = true;
+			nulls[15] = true;
 			nulls[18] = true;
 			nulls[19] = true;
 			nulls[20] = true;
@@ -941,6 +966,7 @@ pg_stat_get_activity(PG_FUNCTION_ARGS)
 			nulls[27] = true;
 			nulls[28] = true;
 			nulls[29] = true;
+			nulls[30] = true;
 		}
 
 		tuplestore_putvalues(tupstore, tupdesc, values, nulls);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index f48f5fb4d9..d24d641fa7 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5228,9 +5228,9 @@
   proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
   proretset => 't', provolatile => 's', proparallel => 'r',
   prorettype => 'record', proargtypes => 'int4',
-  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
-  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
-  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
+  proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,bool,text,numeric,text,bool,text,bool,int4}',
+  proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+  proargnames => '{pid,datid,pid,usesysid,application_name,state,query,individual_query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,sslcompression,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,leader_pid}',
   prosrc => 'pg_stat_get_activity' },
 { oid => '3318',
   descr => 'statistics: information about progress of backends running maintenance command',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 0dfbac46b4..a303c66cdf 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -118,6 +118,13 @@ typedef struct PgStat_TableCounts
 	PgStat_Counter t_blocks_hit;
 } PgStat_TableCounts;
 
+/* Individual Query location and length*/
+typedef struct PgStat_IndividualQuery
+{
+	int         stmt_location;
+	int         stmt_len;
+} PgStat_IndividualQuery;
+
 /* Possible targets for resetting cluster-wide shared values */
 typedef enum PgStat_Shared_Reset_Target
 {
@@ -1128,6 +1135,12 @@ typedef struct PgBackendStatus
 	 */
 	char	   *st_activity_raw;
 
+	/*
+	 * Current individual command location and length
+	 */
+	int stmt_location;
+	int stmt_len;
+
 	/*
 	 * Command progress reporting.  Any command which wishes can advertise
 	 * that it is running by setting st_progress_command,
@@ -1319,7 +1332,7 @@ extern void pgstat_report_checksum_failure(void);
 extern void pgstat_initialize(void);
 extern void pgstat_bestart(void);
 
-extern void pgstat_report_activity(BackendState state, const char *cmd_str);
+extern void pgstat_report_activity(BackendState state, const char *cmd_str, int stmt_location, int stmt_len);
 extern void pgstat_report_tempfile(size_t filesize);
 extern void pgstat_report_appname(const char *appname);
 extern void pgstat_report_xact_timestamp(TimestampTz tstamp);
diff --git a/src/test/modules/worker_spi/worker_spi.c b/src/test/modules/worker_spi/worker_spi.c
index 1c7b17c56f..c4d528f291 100644
--- a/src/test/modules/worker_spi/worker_spi.c
+++ b/src/test/modules/worker_spi/worker_spi.c
@@ -112,7 +112,7 @@ initialize_worker_spi(worktable *table)
 	StartTransactionCommand();
 	SPI_connect();
 	PushActiveSnapshot(GetTransactionSnapshot());
-	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema");
+	pgstat_report_activity(STATE_RUNNING, "initializing worker_spi schema", 0, 0);
 
 	/* XXX could we use CREATE SCHEMA IF NOT EXISTS? */
 	initStringInfo(&buf);
@@ -156,7 +156,7 @@ initialize_worker_spi(worktable *table)
 	SPI_finish();
 	PopActiveSnapshot();
 	CommitTransactionCommand();
-	pgstat_report_activity(STATE_IDLE, NULL);
+	pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 }
 
 void
@@ -262,7 +262,7 @@ worker_spi_main(Datum main_arg)
 		StartTransactionCommand();
 		SPI_connect();
 		PushActiveSnapshot(GetTransactionSnapshot());
-		pgstat_report_activity(STATE_RUNNING, buf.data);
+		pgstat_report_activity(STATE_RUNNING, buf.data, 0, 0);
 
 		/* We can now execute queries via SPI */
 		ret = SPI_execute(buf.data, false, 0);
@@ -292,7 +292,7 @@ worker_spi_main(Datum main_arg)
 		PopActiveSnapshot();
 		CommitTransactionCommand();
 		pgstat_report_stat(false);
-		pgstat_report_activity(STATE_IDLE, NULL);
+		pgstat_report_activity(STATE_IDLE, NULL, 0, 0);
 	}
 
 	proc_exit(1);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2a18dc423e..313993c392 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1761,8 +1761,9 @@ pg_stat_activity| SELECT s.datid,
     s.backend_xid,
     s.backend_xmin,
     s.query,
+    s.individual_query,
     s.backend_type
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      LEFT JOIN pg_database d ON ((s.datid = d.oid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1867,7 +1868,7 @@ pg_stat_gssapi| SELECT s.pid,
     s.gss_auth AS gss_authenticated,
     s.gss_princ AS principal,
     s.gss_enc AS encrypted
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_progress_analyze| SELECT s.pid,
     s.datid,
@@ -2015,7 +2016,7 @@ pg_stat_replication| SELECT s.pid,
     w.sync_priority,
     w.sync_state,
     w.reply_time
-   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
      JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
      LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
 pg_stat_slru| SELECT s.name,
@@ -2037,7 +2038,7 @@ pg_stat_ssl| SELECT s.pid,
     s.ssl_client_dn AS client_dn,
     s.ssl_client_serial AS client_serial,
     s.ssl_issuer_dn AS issuer_dn
-   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
+   FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, individual_query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, sslcompression, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, leader_pid)
   WHERE (s.client_port IS NOT NULL);
 pg_stat_subscription| SELECT su.oid AS subid,
     su.subname,