Copilot commented on code in PR #2029:
URL: https://github.com/apache/cloudberry/pull/2029#discussion_r4038102239
##########
gpcontrib/gp_stats_collector/gp_stats_collector--1.2.sql:
##########
@@ -0,0 +1,159 @@
+/* gp_stats_collector--1.2.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION gp_stats_collector" to load this file. \quit
+
+CREATE SCHEMA gpsc;
+
+CREATE FUNCTION gpsc.__stat_messages_reset_f_on_master()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_stat_messages_reset'
+LANGUAGE C EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__stat_messages_reset_f_on_segments()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_stat_messages_reset'
+LANGUAGE C EXECUTE ON ALL SEGMENTS;
+
+CREATE FUNCTION gpsc.stat_messages_reset()
+RETURNS SETOF void
+AS
+$$
+ SELECT gpsc.__stat_messages_reset_f_on_master();
+ SELECT gpsc.__stat_messages_reset_f_on_segments();
+$$
+LANGUAGE SQL EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__stat_messages_f_on_master()
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'gpsc_stat_messages'
+LANGUAGE C STRICT VOLATILE EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__stat_messages_f_on_segments()
+RETURNS SETOF record
+AS 'MODULE_PATHNAME', 'gpsc_stat_messages'
+LANGUAGE C STRICT VOLATILE EXECUTE ON ALL SEGMENTS;
+
+CREATE VIEW gpsc.stat_messages AS
+ SELECT C.*
+ FROM gpsc.__stat_messages_f_on_master() as C (
+ segid int,
+ total_messages bigint,
+ send_failures bigint,
+ connection_failures bigint,
+ other_errors bigint,
+ max_message_size int
+ )
+ UNION ALL
+ SELECT C.*
+ FROM gpsc.__stat_messages_f_on_segments() as C (
+ segid int,
+ total_messages bigint,
+ send_failures bigint,
+ connection_failures bigint,
+ other_errors bigint,
+ max_message_size int
+ )
+ORDER BY segid;
+
+CREATE FUNCTION gpsc.__init_log_on_master()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_init_log'
+LANGUAGE C STRICT VOLATILE EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__init_log_on_segments()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_init_log'
+LANGUAGE C STRICT VOLATILE EXECUTE ON ALL SEGMENTS;
+
+-- Creates log table inside gpsc schema.
+SELECT gpsc.__init_log_on_master();
+SELECT gpsc.__init_log_on_segments();
+
+CREATE VIEW gpsc.log AS
+ SELECT * FROM gpsc.__log -- master
+ UNION ALL
+ SELECT * FROM gp_dist_random('gpsc.__log') -- segments
+ORDER BY tmid, ssid, ccnt;
+
+CREATE FUNCTION gpsc.__truncate_log_on_master()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_truncate_log'
+LANGUAGE C STRICT VOLATILE EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__truncate_log_on_segments()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_truncate_log'
+LANGUAGE C STRICT VOLATILE EXECUTE ON ALL SEGMENTS;
+
+CREATE FUNCTION gpsc.truncate_log()
+RETURNS SETOF void AS $$
+BEGIN
+ PERFORM gpsc.__truncate_log_on_master();
+ PERFORM gpsc.__truncate_log_on_segments();
+END;
+$$ LANGUAGE plpgsql VOLATILE;
+
+CREATE FUNCTION gpsc.__test_uds_start_server(path text)
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_test_uds_start_server'
+LANGUAGE C STRICT EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__test_uds_receive(timeout_ms int DEFAULT 2000)
+RETURNS SETOF bigint
+AS 'MODULE_PATHNAME', 'gpsc_test_uds_receive'
+LANGUAGE C STRICT EXECUTE ON COORDINATOR;
+
+CREATE FUNCTION gpsc.__test_uds_stop_server()
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'gpsc_test_uds_stop_server'
+LANGUAGE C EXECUTE ON COORDINATOR;
+
+-- ---------------------------------------------------------------------------
+-- 1.2: pg_query_state per-node runtime collection (push to yagpcc via UDS)
+-- ---------------------------------------------------------------------------
+
+-- Compact (segid, pid) identifier for a QE backend running on a segment.
+-- Matches the C gp_segment_pid struct used by the pg_query_state signal layer.
+CREATE TYPE gpsc.gp_segment_pid AS (
+ segid int,
+ pid int
+);
+
+-- pg_query_state(pid): trigger runtime per-node collection for the query
+-- running on backend `pid`. Fans QueryStatePollReason out to every QE via
+-- cbdb_mpp_query_state; each matching QE walks its plan tree and pushes a
+-- per-node batch to its local yagpcc over UDS. Fire-and-forget: returns void.
+CREATE FUNCTION gpsc.pg_query_state(pid int, trace_id bytea)
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'pg_query_state'
+LANGUAGE C VOLATILE EXECUTE ON COORDINATOR;
Review Comment:
These C functions are not declared `STRICT`. A caller can pass `NULL` for
`trace_id` (or for the internal array), causing
`PG_GETARG_BYTEA_P`/`PG_GETARG_ARRAYTYPE_P` and the subsequent
`VARSIZE_ANY_EXHDR` access to dereference an absent argument instead of
returning a controlled error. Mark the trace-bearing declarations `STRICT` in
both the fresh-install and 1.1-to-1.2 scripts, or check `PG_ARGISNULL()` before
each `PG_GETARG_*`.
##########
src/backend/storage/ipc/procsignal.c:
##########
@@ -101,12 +101,20 @@ typedef struct
#define BARRIER_CLEAR_BIT(flags, type) \
((flags) &= ~(((uint32) 1) << (uint32) (type)))
+#define IsCustomProcSignalReason(reason) \
+ ((reason) >= PROCSIG_CUSTOM_1 && (reason) <= PROCSIG_CUSTOM_N)
+
+static bool CustomSignalPendings[NUM_CUSTOM_PROCSIGNALS];
+static bool CustomSignalProcessing[NUM_CUSTOM_PROCSIGNALS];
+static ProcSignalHandler_type CustomInterruptHandlers[NUM_CUSTOM_PROCSIGNALS];
Review Comment:
`CustomSignalPendings` is written from `procsignal_sigusr1_handler`, but it
is a plain `bool` array rather than `volatile sig_atomic_t`. The compiler is
therefore not required to observe a signal-handler update when
`CheckAndHandleCustomSignals()` reads it, so custom notifications can be lost
under optimization. Use a `volatile sig_atomic_t` pending array, matching the
shared `pss_signalFlags` contract.
##########
gpcontrib/gp_stats_collector/src/pg_query_state/signal_handler.c:
##########
@@ -0,0 +1,1012 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * signal_handler.c
+ * Custom signal handlers and plan-tree walker for pg_query_state.
+ *
+ * This module implements the two custom ProcSignal handlers registered by
+ * pg_qs_init():
+ *
+ * SendQueryState() -- fired when QueryStatePollReason is received.
+ * Walks the active plan tree, collects per-node
stats,
+ * logs them, then pushes the whole snapshot to the
+ * yagpcc UDS sink (and, on the coordinator, the
+ * deparsed plan document).
+ * SendCdbComponents() -- fired when BackendInfoPollReason is received (QD
only).
+ * Sends the list of active QE (segid, pid) pairs.
+ *
+ * Also contains:
+ * qs_planstate_walker() -- recursive plan-tree traversal helper.
+ * qs_get_node_stats() -- per-node stat collection callback.
+ * qs_debug_node_stats() -- LOG-level dump of a collected stat list.
+ * qs_debug_node_sample() -- LOG-level dump of a single GpscNodeSample.
+ * send_msg_by_parts() -- chunked shm_mq send helper.
+ *
+ * Portions derived from pg_query_state
+ * (https://github.com/postgrespro/pg_query_state), under the PostgreSQL
+ * License:
+ * Portions Copyright (c) 2016-2025, Postgres Professional
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_stats_collector/src/pg_query_state/signal_handler.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include <unistd.h>
+
+#include "pg_query_state.h"
+#include "PlanNodeEmitter.h"
+
+#include "access/xact.h"
+#include "cdb/cdbexplain.h"
+#include "cdb/cdbutil.h"
+#include "cdb/cdbvars.h"
+#include "libpq-fe.h"
+#include "cdb/cdbconn.h"
+#include "commands/explain.h"
+#include "executor/executor.h"
+#include "miscadmin.h"
+#include "nodes/execnodes.h"
+#include "nodes/plannodes.h"
+#include "pgstat.h"
+#include "parser/parsetree.h"
+#include "storage/bufmgr.h"
+#include "storage/lock.h"
+#include "utils/builtins.h"
+#include "utils/memutils.h"
+#include "utils/resowner.h"
+#include "utils/rel.h"
+#include "utils/timestamp.h"
+#include "utils/hsearch.h"
+#include "utils/lsyscache.h"
+#include "libpq/pqmq.h"
+
+/*
+ * Identity of the most recent coordinator plan-doc push, used to rate-limit
+ * SetQueryPlanReq: SendQueryState() re-sends the deparsed plan only when the
+ * query key changes or PLAN_DOC_RESEND_INTERVAL_MS has elapsed.
+ */
+static struct
+{
+ int32_t tmid;
+ int32_t ssid;
+ int32_t ccnt;
+ TimestampTz at;
+} last_sent_query_key;
+
+typedef struct NodeRollState
+{
+ int32_t plan_node_id;
+ double prev_ntuples_sum;
+ TimestampTz prev_executed_at;
+ TimestampTz first_executed_at;
+ int32_t relation_oid;
+ char relation_name[MAX_RELNAME_LEN];
+} NodeRollState;
+
+static HTAB *node_roll_htab = NULL;
+
+static void ensure_node_roll_htab(void)
+{
+ HASHCTL ctl;
+
+ if (node_roll_htab)
+ {
+ return;
+ }
+
+ memset(&ctl, 0, sizeof(ctl));
+ ctl.keysize = sizeof(int);
+ ctl.entrysize = sizeof(NodeRollState);
+ ctl.hcxt = TopMemoryContext;
+ node_roll_htab = hash_create("gpsc_per_node_roll_state",
+ 64, &ctl,
HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
+}
+
+void gpsc_reset_node_roll_state(void)
+{
+ if (node_roll_htab)
+ {
+ hash_destroy(node_roll_htab);
+ node_roll_htab = NULL;
+ }
+}
+
+/*
+ * qs_reporting_segid -- segid this backend stamps on the samples it emits.
+ *
+ * GpIdentity.segindex is -1 both here and on the QD; an executor role on the
+ * coordinator host means the entry-db QE, which needs a segid of its own.
+ */
+static int32
+qs_reporting_segid(void)
+{
+ if (Gp_role == GP_ROLE_EXECUTE && GpIdentity.segindex < 0)
+ return GPSC_SEGID_ENTRY_DB;
+
+ return GpIdentity.segindex;
+}
+
+/*
+ * shm_mq_send_nonblocking -- attempt to send nbytes through mqh up to
+ * `attempts` times, sleeping WRITING_DELAY µs between retries.
+ *
+ * Returns MSG_BY_PARTS_FAILED immediately on SHM_MQ_DETACHED; retries on
+ * SHM_MQ_WOULD_BLOCK.
+ */
+static msg_by_parts_result
+shm_mq_send_nonblocking(shm_mq_handle *mqh, Size nbytes,
+ const void *data, Size attempts)
+{
+ int i;
+ shm_mq_result res;
+
+ for (i = 0; i < (int) attempts; i++)
+ {
+#if PG_VERSION_NUM < 150000
+ res = shm_mq_send(mqh, nbytes, data, true);
+#else
+ res = shm_mq_send(mqh, nbytes, data, true, true);
+#endif
+
+ if (res == SHM_MQ_SUCCESS)
+ break;
+ else if (res == SHM_MQ_DETACHED)
+ return MSG_BY_PARTS_FAILED;
+
+ /* SHM_MQ_WOULD_BLOCK -- back off briefly and retry. */
+ pg_usleep(WRITING_DELAY);
+ }
+
+ if (i == (int) attempts)
+ return MSG_BY_PARTS_FAILED;
+
+ return MSG_BY_PARTS_SUCCEEDED;
+}
+
+/*
+ * send_msg_by_parts -- transmit an arbitrarily large buffer through mqh.
+ *
+ * The wire protocol is: first send a Size value announcing the total payload
+ * length, then send the payload itself in chunks of at most MSG_MAX_SIZE
+ * bytes. The receiver must use receive_msg_by_parts() (in pg_query_state.c)
+ * to reassemble the chunks.
+ *
+ * Parameters:
+ * mqh -- attached shm_mq handle (sender side)
+ * nbytes -- total payload size
+ * data -- pointer to the payload
+ *
+ * Returns MSG_BY_PARTS_SUCCEEDED on success, MSG_BY_PARTS_FAILED otherwise.
+ */
+static msg_by_parts_result
+send_msg_by_parts(shm_mq_handle *mqh, Size nbytes, const void *data)
+{
+ int offset;
+ int bytes_left;
+ int bytes_send;
+
+ /* Announce total length. */
+ if (shm_mq_send_nonblocking(mqh, sizeof(Size), &nbytes,
+
NUM_OF_ATTEMPTS) == MSG_BY_PARTS_FAILED)
+ return MSG_BY_PARTS_FAILED;
+
+ /* Send payload in chunks. */
+ for (offset = 0; offset < (int) nbytes; offset += bytes_send)
+ {
+ bytes_left = nbytes - offset;
+ bytes_send = (bytes_left < MSG_MAX_SIZE) ? bytes_left :
MSG_MAX_SIZE;
+ if (shm_mq_send_nonblocking(mqh, bytes_send,
+
&(((unsigned char *) data)[offset]),
+
NUM_OF_ATTEMPTS) == MSG_BY_PARTS_FAILED)
+ return MSG_BY_PARTS_FAILED;
+ }
+
+ return MSG_BY_PARTS_SUCCEEDED;
+}
+
+/*
+ * qs_planstate_walker -- depth-first traversal of a PlanState tree.
+ *
+ * Visits every node in the tree rooted at `planstate`, calling `executor`
+ * on each node before recursing. Handles all node types that have child
+ * plan states (Append, MergeAppend, Sequence, BitmapAnd/Or, SubqueryScan,
+ * CustomScan, init-plans, and sub-plans) -- keep the switch below in step with
+ * the "special child plans" switch in ExplainNode().
+ *
+ * Parameters:
+ * planstate -- root of the subtree to walk (NULL is a no-op)
+ * executor -- callback invoked for each node
+ * qs_walker_ctx -- context threaded through all callbacks
+ * depth -- current recursion depth (for stack-depth checks)
+ */
+static void
+qs_planstate_walker(PlanState *planstate,
+ qs_planstate_walker_callback executor,
+ QsWalkerContext *qs_walker_ctx,
+ int depth)
+{
+ int32 saved_parent_plan_node_id;
+ int32 saved_slice_id;
+ Plan *plan;
+ ListCell *lc;
+
+ if (planstate == NULL)
+ return;
+
+ check_stack_depth();
+
+ plan = planstate->plan;
+
+ /*
+ * A Motion opens a new slice, and the node itself belongs to the
sending
+ * side -- the same attribution ExplainNode() uses. Switch before
sampling
+ * so the Motion is reported under its own slice, not its parent's.
+ */
+ saved_slice_id = qs_walker_ctx->slice_id;
+ if (IsA(plan, Motion))
+ {
+ Motion *motion = (Motion *) plan;
+ SliceTable *sliceTable = planstate->state->es_sliceTable;
+
+ if (sliceTable && motion->motionID >= 0 &&
+ motion->motionID < sliceTable->numSlices)
+ qs_walker_ctx->slice_id =
sliceTable->slices[motion->motionID].sliceIndex;
+ }
+
+ executor(planstate, qs_walker_ctx);
+ saved_parent_plan_node_id = qs_walker_ctx->parent_plan_node_id;
+ qs_walker_ctx->parent_plan_node_id = plan->plan_node_id;
+
+ /* initPlans */
+ foreach(lc, planstate->initPlan)
+ {
+ SubPlanState *sps = lfirst_node(SubPlanState, lc);
+ qs_planstate_walker(sps->planstate, executor, qs_walker_ctx,
depth + 1);
+ }
+
+ /* Left and right children. */
+ qs_planstate_walker(outerPlanState(planstate), executor, qs_walker_ctx,
+ depth + 1);
+ qs_planstate_walker(innerPlanState(planstate), executor, qs_walker_ctx,
+ depth + 1);
+
+ /* Type-specific child plans. */
+ switch (nodeTag(plan))
+ {
+ case T_Append:
+ {
+ AppendState *as = (AppendState *) planstate;
+ for (int i = 0; i < as->as_nplans; i++)
+ qs_planstate_walker(as->appendplans[i],
executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ }
+ case T_MergeAppend:
+ {
+ MergeAppendState *ms = (MergeAppendState *) planstate;
+ for (int i = 0; i < ms->ms_nplans; i++)
+ qs_planstate_walker(ms->mergeplans[i], executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ }
+ case T_Sequence:
+ {
+ SequenceState *ss = (SequenceState *) planstate;
+ for (int i = 0; i < ss->numSubplans; i++)
+ qs_planstate_walker(ss->subplans[i], executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ }
+ case T_BitmapAnd:
+ {
+ BitmapAndState *bas = (BitmapAndState *) planstate;
+ for (int i = 0; i < bas->nplans; i++)
+ qs_planstate_walker(bas->bitmapplans[i],
executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ }
+ case T_BitmapOr:
+ {
+ BitmapOrState *bos = (BitmapOrState *) planstate;
+ for (int i = 0; i < bos->nplans; i++)
+ qs_planstate_walker(bos->bitmapplans[i],
executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ }
+ case T_SubqueryScan:
+ qs_planstate_walker(((SubqueryScanState *)
planstate)->subplan,
+ executor,
qs_walker_ctx, depth + 1);
+ break;
+ case T_CustomScan:
+ foreach(lc, ((CustomScanState *) planstate)->custom_ps)
+ qs_planstate_walker((PlanState *) lfirst(lc),
executor,
+
qs_walker_ctx, depth + 1);
+ break;
+ default:
+ break;
+ }
+
+ /* subPlans */
+ foreach(lc, planstate->subPlan)
+ {
+ SubPlanState *sps = lfirst_node(SubPlanState, lc);
+ qs_planstate_walker(sps->planstate, executor, qs_walker_ctx,
depth + 1);
+ }
+
+ qs_walker_ctx->parent_plan_node_id = saved_parent_plan_node_id;
+ qs_walker_ctx->slice_id = saved_slice_id;
+}
+
+/*
+ * qs_get_node_stats -- walker callback that snapshots one plan node.
+ *
+ * Allocates a GpscNodeSample in the current memory context, fills it from
+ * planstate->instrument (if available), and appends it to
+ * qs_walker_ctx->per_node_stats.
+ *
+ * Parameters:
+ * planstate -- the plan node being sampled
+ * qs_walker_ctx -- walker context; per_node_stats is extended in-place
+ */
+static void
+qs_get_node_stats(PlanState *planstate, QsWalkerContext *qs_walker_ctx)
+{
+ GpscNodeSample *nodestat =
+ (GpscNodeSample *) palloc0(sizeof(GpscNodeSample));
+
+ /* Identity fields. */
+ nodestat->ssid = gp_session_id;
+ nodestat->tmid = qs_walker_ctx->tmid;
+ nodestat->ccnt = gp_command_count;
+
+ /* Plan-tree position. */
+ nodestat->plan_node_id = planstate->plan->plan_node_id;
+ nodestat->parent_plan_node_id = qs_walker_ctx->parent_plan_node_id;
+ nodestat->node_tag = nodeTag(planstate->plan);
+ nodestat->slice_id = qs_walker_ctx->slice_id;
+ nodestat->segindex = qs_reporting_segid();
+ nodestat->dbid = GpIdentity.dbid;
+ nodestat->pid = MyProcPid;
+
+ /* Planner estimate. */
+ nodestat->plan_rows = planstate->plan->plan_rows;
+
+ /* Runtime instrumentation (may be NULL for non-instrumented nodes). */
+ if (planstate->instrument)
+ {
+ Instrumentation *instr = planstate->instrument;
+ double eff_nloops;
+
+ if (qs_walker_ctx->finalize)
+ {
+ InstrEndLoop(instr);
+ }
+
+ eff_nloops = instr->nloops;
+ if (!qs_walker_ctx->finalize && instr->eof)
+ eff_nloops += 1;
+
+ nodestat->ntuples = instr->ntuples + instr->tuplecount; /*
include in-progress loop */
+ nodestat->tuplecount = instr->tuplecount;
+ nodestat->nloops = eff_nloops;
+ nodestat->startup = instr->startup;
+ nodestat->total = instr->total;
+ nodestat->firsttuple = instr->firsttuple;
+
+ nodestat->shared_blks_hit = instr->bufusage.shared_blks_hit;
+ nodestat->shared_blks_read = instr->bufusage.shared_blks_read;
+
+ /*
+ * eof lets a consumer tell a node that has finished producing
(running
+ * but exhausted for this cycle) from one still actively
pulling.
+ */
+ nodestat->eof = instr->eof;
+
+ if (instr->running && !instr->eof)
+ nodestat->node_status = QS_NODE_STATUS_EXECUTING;
+ else if (eff_nloops > 0)
+ nodestat->node_status = QS_NODE_STATUS_FINISHED;
+ else
+ nodestat->node_status = QS_NODE_STATUS_INITIALIZED;
+
+ nodestat->workfile_created = instr->workfileCreated;
+ nodestat->workmem_used = (int64_t) instr->workmemused;
+ nodestat->workmem_wanted = (int64_t) instr->workmemwanted;
+ }
+ else
+ {
+ nodestat->node_status = QS_NODE_STATUS_INITIALIZED;
+ }
+
+ qs_walker_ctx->per_node_stats =
+ lappend(qs_walker_ctx->per_node_stats, nodestat);
+
+ {
+ TimestampTz ts_now = qs_walker_ctx->ts_now;
+ double cur_sum = nodestat->ntuples;
+ bool found;
+ NodeRollState *rs;
+
+ rs = (NodeRollState *) hash_search(node_roll_htab,
+ &nodestat->plan_node_id, HASH_ENTER, &found);
+
+ if (found)
+ {
+ double dt = (double) (ts_now - rs->prev_executed_at) /
USECS_PER_SEC;
+ nodestat->ntuples_delta = cur_sum -
rs->prev_ntuples_sum;
+ nodestat->tuples_per_sec = (dt > 0) ?
nodestat->ntuples_delta / dt : 0;
+ nodestat->time_since_init_sec = (double) (ts_now -
rs->first_executed_at) / USECS_PER_SEC;
+
+ /* Relation identity is invariant per plan node: reuse
the cache. */
+ nodestat->relation_oid = rs->relation_oid;
+ strlcpy(nodestat->relation_name, rs->relation_name,
MAX_RELNAME_LEN);
+ }
+ else
+ {
+ Index rti = 0;
+
+ nodestat->ntuples_delta = cur_sum;
+ nodestat->tuples_per_sec = 0;
+ nodestat->time_since_init_sec = 0;
+ rs->first_executed_at = ts_now;
+
+ switch (nodeTag(planstate->plan))
+ {
+ case T_SeqScan:
+ case T_DynamicSeqScan:
+ case T_SampleScan:
+ case T_IndexScan:
+ case T_DynamicIndexScan:
+ case T_DynamicIndexOnlyScan:
+ case T_IndexOnlyScan:
+ case T_BitmapHeapScan:
+ case T_DynamicBitmapHeapScan:
+ case T_TidScan:
+ case T_TidRangeScan:
+ case T_ForeignScan:
+ case T_DynamicForeignScan:
+ case T_CustomScan:
+ rti = ((Scan *)
planstate->plan)->scanrelid;
+ break;
+ case T_ModifyTable:
+ rti = ((ModifyTable *)
planstate->plan)->nominalRelation;
+ break;
+ default:
+ break;
+ }
+
+ if (rti > 0 && planstate->state)
+ {
+ List *rtable = planstate->state->es_range_table;
+ if (rti <= (Index) list_length(rtable))
+ {
+ RangeTblEntry *rte = rt_fetch(rti,
rtable);
+ if (rte->rtekind == RTE_RELATION)
+ {
+ char *relname;
+ nodestat->relation_oid =
(int32_t) rte->relid;
+ relname =
get_rel_name(rte->relid);
+ if (relname)
+ {
+
strlcpy(nodestat->relation_name, relname, MAX_RELNAME_LEN);
+ pfree(relname);
+ }
+ }
+ }
+ }
+
+ /* Cache the resolved identity for subsequent polls. */
+ rs->relation_oid = nodestat->relation_oid;
+ strlcpy(rs->relation_name, nodestat->relation_name,
MAX_RELNAME_LEN);
+ }
+
+ nodestat->stalled = (nodestat->ntuples_delta == 0
+ && nodestat->node_status ==
QS_NODE_STATUS_EXECUTING
+ && !nodestat->eof);
+ rs->prev_ntuples_sum = cur_sum;
+ rs->prev_executed_at = ts_now;
+ }
+}
+
+/*
+ * qs_debug_node_sample -- emit a single GpscNodeSample to the PostgreSQL LOG.
+ *
+ * Intended for development and integration testing. In production deployments
+ * this will produce a large number of log lines; suppress with
log_min_messages.
+ */
+static void
+qs_debug_node_sample(GpscNodeSample *s)
+{
+ elog(DEBUG1,
+ "GpscNodeSample: "
+ "plan_node_id=%d parent=%d node_tag=%d "
+ "slice_id=%d segindex=%d "
+ "tmid=%d ssid=%d ccnt=%d "
+ "plan_rows=%.0f "
+ "ntuples=%.0f tuplecount=%.0f nloops=%.0f "
+ "startup=%f total=%f firsttuple=%f "
+ "shared_blks_hit=" UINT64_FORMAT " shared_blks_read="
UINT64_FORMAT " "
+ "workfile_created=%d workmem_used=" INT64_FORMAT
+ " workmem_wanted=" INT64_FORMAT " "
+ "node_status=%d",
+ s->plan_node_id, s->parent_plan_node_id, s->node_tag,
+ s->slice_id, s->segindex,
+ s->tmid, s->ssid, s->ccnt,
+ s->plan_rows,
+ s->ntuples, s->tuplecount, s->nloops,
+ s->startup, s->total, s->firsttuple,
+ (uint64) s->shared_blks_hit, (uint64) s->shared_blks_read,
+ (int) s->workfile_created, (int64) s->workmem_used, (int64)
s->workmem_wanted,
+ (int) s->node_status);
+}
+
+/*
+ * qs_debug_node_stats -- emit all nodes in per_node_stats to the PostgreSQL
LOG.
+ *
+ * Logs a summary line followed by one line per node via
qs_debug_node_sample().
+ */
+static void
+qs_debug_node_stats(List *per_node_stats)
+{
+ ListCell *lc;
+ int i = 0;
+
+ if (!message_level_is_interesting(DEBUG1))
+ return;
+
+ elog(DEBUG1, "GpscNodeSample list: %d nodes",
list_length(per_node_stats));
+ foreach(lc, per_node_stats)
+ {
+ GpscNodeSample *s = (GpscNodeSample *) lfirst(lc);
+ elog(DEBUG1, "--- node[%d] ---", i++);
+ qs_debug_node_sample(s);
+ }
+}
+
+/*
+ * runtime_explain -- snapshot the active query's plan tree.
+ *
+ * Retrieves the top-most QueryDesc from QueryDescStack, walks its planstate
+ * tree with qs_get_node_stats(), and returns the resulting List of
+ * GpscNodeSample pointers.
+ *
+ * Callers must ensure QueryDescStack is non-empty before calling this.
+ */
+static List *
+runtime_explain(TimestampTz ts_now)
+{
+ QsWalkerContext *qs_walker_ctx =
+ (QsWalkerContext *) palloc0(sizeof(QsWalkerContext));
+ QueryDesc *queryDesc;
+
+ Assert(list_length(QueryDescStack) > 0);
+ queryDesc = get_toppest_query();
+ qs_walker_ctx->ts_now = ts_now;
+ qs_walker_ctx->parent_plan_node_id = GPSC_NO_PARENT_PLAN_NODE_ID;
+ qs_walker_ctx->slice_id = queryDesc->estate
+ ? LocallyExecutingSliceIndex(queryDesc->estate)
+ : currentSliceId;
+ gp_gettmid(&qs_walker_ctx->tmid);
+ ensure_node_roll_htab();
+ qs_planstate_walker(queryDesc->planstate, qs_get_node_stats,
+ qs_walker_ctx, 0);
+ return qs_walker_ctx->per_node_stats;
+}
+
+/*
+ * emit_node_batch -- push a whole plan-tree snapshot as one
SetPerNodeBatchReq.
+ *
+ * Flattens the List<GpscNodeSample *> into a contiguous array and hands it to
+ * the C++ emitter, which opens a single UDS connection for the whole backend
+ * instead of one connection per node. A NULL or empty list is a no-op.
+ *
+ * The caller is responsible for calling gpsc_qs_sync_config() beforehand.
+ */
+static void
+emit_node_batch(List *per_node_stats, const char *trace_id)
+{
+ GpscNodeSample **arr;
+ ListCell *lc;
+ int n = list_length(per_node_stats);
+ int i = 0;
+
+ if (n == 0)
+ return;
+
+ arr = (GpscNodeSample **) palloc(n * sizeof(GpscNodeSample *));
+ foreach(lc, per_node_stats)
+ arr[i++] = (GpscNodeSample *) lfirst(lc);
+
+ gpsc_emit_node_batch(arr, n, trace_id);
+}
+
+/*
+ * build_plan_doc -- render the active query's plan via ExplainPrintPlan.
+ *
+ * Produces the full deparsed plan document (expressions, costs, Settings) in
+ * the requested ExplainFormat. ExplainBeginOutput/ExplainEndOutput and the
+ * enclosing "Query" group frame the output so JSON/XML/YAML come out
+ * well-formed: ExplainPrintPlan on its own renders only the inner "Plan"
+ * property, so without the group the non-text formats are an unwrapped
+ * fragment no parser accepts. The framing lives here, outside
+ * ExplainPrintPlan, so that function is left untouched.
+ *
+ * Returns a palloc'd string in the current context, or NULL when there is
+ * nothing to render. The coordinator (QD) restriction is enforced here rather
+ * than left to the caller: a QE only instantiates the PlanStates of its own
+ * slice, so the document it produced would be a partial tree that the
collection
+ * has no use for -- the QD's copy is the whole plan.
+ */
+static char *
+build_plan_doc(QueryDesc *queryDesc, ExplainFormat format)
+{
+ ExplainState *es;
+
+ if (queryDesc == NULL || Gp_role != GP_ROLE_DISPATCH)
+ return NULL;
+
+ HOLD_INTERRUPTS();
Review Comment:
`SendQueryState()` already holds interrupts before calling this helper. If
`ExplainPrintPlan()` raises an error, the non-local jump skips
`RESUME_INTERRUPTS()` below, leaving this extra hold count installed; the outer
catch only balances the caller's hold, so later `CHECK_FOR_INTERRUPTS()` calls
in that backend can remain suppressed. Remove this nested hold pair or protect
it with a `PG_FINALLY` cleanup.
##########
gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.c:
##########
@@ -0,0 +1,1192 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * pg_query_state.c
+ * Core of the pg_query_state signal-dispatch layer.
+ *
+ * This module provides:
+ * - Shared-memory setup (shm_toc segment with params, mq, mq_req_id).
+ * - Custom ProcSignal registrations for three signals:
+ * QueryStatePollReason -> SendQueryState()
+ * BackendInfoPollReason -> SendCdbComponents()
+ * - GUC variables: pg_query_state.enable / enable_timing / enable_buffers.
+ * - Executor hooks (start/run/finish/end), registered by this module itself,
+ * that maintain the QueryDescStack and enable instrumentation on the
+ * top-level query.
+ * - A requestor-side helper: shm_mq_receive_with_timeout().
+ *
+ * Per-node stats are pushed to the yagpcc UDS sink on demand, when a backend
is
+ * signalled to report its live query state; see signal_handler.c.
+ *
+ * Portions derived from pg_query_state
+ * (https://github.com/postgrespro/pg_query_state), under the PostgreSQL
+ * License:
+ * Portions Copyright (c) 2016-2025, Postgres Professional
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "pg_query_state.h"
+#include "PlanNodeEmitter.h"
+
+#include "access/htup_details.h"
+#include "access/xact.h"
+#include "catalog/pg_type.h"
+#include "cdb/cdbdispatchresult.h"
+#include "cdb/cdbdisp_query.h"
+#include "cdb/cdbexplain.h"
+#include "cdb/cdbvars.h"
+#include "executor/execParallel.h"
+#include "executor/executor.h"
+#include "fmgr.h"
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "nodes/nodeFuncs.h"
+#include "nodes/print.h"
+#include "parser/analyze.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "storage/ipc.h"
+#include "storage/s_lock.h"
+#include "storage/spin.h"
+#include "storage/procarray.h"
+#include "storage/procsignal.h"
+#include "storage/shm_toc.h"
+#include "utils/builtins.h"
+#include "utils/guc.h"
+#include "utils/timestamp.h"
+#include "utils/lsyscache.h"
+#include "utils/portal.h"
+#include "utils/typcache.h"
+
+/* GUC variables */
+/* Master switch: disabling this suppresses all stat collection. */
+bool pg_qs_enable = true;
+
+/* Collect timing (wall-clock) data in addition to row counts. */
+bool pg_qs_timing = true;
+
+/* Collect buffer usage via Instrumentation.bufusage. */
+bool pg_qs_buffers = true;
+
+/*
+ * Rolling counter incremented for every QueryDesc pushed onto the stack.
+ * Used to generate synthetic queryId values for statements lacking one.
+ */
+static int qs_push_count = 0;
+
+/* Saved hook pointer for chaining shmem_startup callbacks. */
+static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+
+/* Saved hook pointers for chaining the executor callbacks. */
+static ExecutorStart_hook_type prev_ExecutorStart_hook = NULL;
+static ExecutorRun_hook_type prev_ExecutorRun_hook = NULL;
+static ExecutorFinish_hook_type prev_ExecutorFinish_hook = NULL;
+static ExecutorEnd_hook_type prev_ExecutorEnd_hook = NULL;
+
+/* Whether pg_qs_shmem_startup has completed successfully. */
+static bool module_initialized = false;
+
+/*
+ * Monotonically increasing request counter on the requestor side.
+ * Compared against *mq_req_id in the reply to detect stale responses.
+ */
+static int reqid = 0;
+
+/* Shared-memory variables (pointers into the shm_toc segment) */
+/* Table of contents anchoring the whole shared segment. */
+static shm_toc *toc = NULL;
+
+/*
+ * Signal parameters written by the requestor and read by the handler.
+ * Slot 0 in the toc.
+ */
+pg_qs_params *params = NULL;
+
+/*
+ * Raw shared memory queue used to return data from the handler.
+ * Slot 1 in the toc.
+ */
+shm_mq *mq = NULL;
+
+/*
+ * Shared request-id counter. The requestor increments it before sending a
+ * signal; the handler echoes it back so the requestor can detect stale
+ * replies. Slot 2 in the toc.
+ */
+uint32 *mq_req_id = NULL;
+
+/*
+ * Per-backend trace_id slots (toc key 3), indexed by BackendId. The
dispatcher
+ * stamps the target's slot before signalling; the signaled backend reads its
+ * own slot to key the batch it pushes. See the header for the full rationale.
+ */
+char (*qs_trace_slots)[GPSC_TRACE_ID_LEN] = NULL;
+
+/* Global signal-reason handles (set during pg_qs_init) */
+List *QueryDescStack = NIL;
+
+ProcSignalReason QueryStatePollReason = INVALID_PROCSIGNAL;
+ProcSignalReason BackendInfoPollReason = INVALID_PROCSIGNAL;
+
+/* Forward declarations for module-private helpers */
+static Size pg_qs_shmem_size(void);
+static void pg_qs_shmem_startup(void);
+static void push_query(QueryDesc *queryDesc);
+static void pg_qs_pop_query(void);
+static bool filter_query(QueryDesc *queryDesc);
+static void pg_qs_executor_start(QueryDesc *queryDesc, int eflags);
+static void pg_qs_executor_run(QueryDesc *queryDesc, ScanDirection direction,
+ uint64 count, bool
execute_once);
+static void pg_qs_executor_finish(QueryDesc *queryDesc);
+static void pg_qs_executor_end(QueryDesc *queryDesc);
+static shm_mq_result shm_mq_receive_with_timeout(shm_mq_handle *mqh, Size
*nbytesp,
+
void **datap, int64 timeout);
+static List *get_query_backend_info(ArrayType *array);
+static shm_mq_result receive_msg_by_parts(shm_mq_handle *mqh, Size *total,
+
void **datap, int64 timeout,
+
int *rc, bool nowait);
+static PG_QS_RequestResult GetRemoteBackendInfo(PGPROC *proc, List **result);
+static void CollectQEQueryState(List *backendInfo, bytea *trace_id);
+static void SignalEntryDbBackends(List *backendInfo, bytea *trace_id);
+static bool is_querystack_empty(void);
+static PG_QS_RequestResult qs_fetch_backend_info(PGPROC *proc, List
**backend_info);
+
+#if PG_VERSION_NUM >= 150000
+static shmem_request_hook_type prev_shmem_request_hook = NULL;
+static void pg_qs_shmem_request(void);
+#endif
+
+/*
+ * pg_qs_shmem_size -- compute the size of the shared memory segment.
+ *
+ * The segment holds four objects at fixed toc keys:
+ * key 0: pg_qs_params
+ * key 1: message queue of QUEUE_SIZE bytes
+ * key 2: uint32 request-id counter
+ * key 3: per-backend trace_id slots, char[GPSC_TRACE_ID_LEN] ×
(MaxBackends+1)
+ */
+static Size
+pg_qs_shmem_size(void)
+{
+ shm_toc_estimator e;
+ Size size;
+ int nkeys = 4;
+
+ shm_toc_initialize_estimator(&e);
+ shm_toc_estimate_chunk(&e, sizeof(pg_qs_params));
+ shm_toc_estimate_chunk(&e, (Size) QUEUE_SIZE);
+ shm_toc_estimate_chunk(&e, sizeof(uint32));
+ shm_toc_estimate_chunk(&e, (Size) GPSC_TRACE_ID_LEN * (MaxBackends +
1));
+ shm_toc_estimate_keys(&e, nkeys);
+ size = shm_toc_estimate(&e);
+ return size;
+}
+
+/*
+ * pg_qs_shmem_startup -- attach to (or initialize) the shared segment.
+ *
+ * Called from the shmem_startup_hook chain after shared memory is mapped.
+ * On first call (found == false) it initialises all sub-structures.
+ * On subsequent calls it just re-attaches the toc pointers.
+ */
+static void
+pg_qs_shmem_startup(void)
+{
+ bool found;
+ Size shmem_size = pg_qs_shmem_size();
+ void *shmem;
+ int num_toc = 0;
+
+ LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
+ shmem = ShmemInitStruct("pg_query_state", shmem_size, &found);
+ if (!found)
+ {
+ toc = shm_toc_create(PG_QS_MODULE_KEY, shmem, shmem_size);
+
+ params = shm_toc_allocate(toc, sizeof(pg_qs_params));
+ shm_toc_insert(toc, num_toc++, params);
+
+ mq = shm_toc_allocate(toc, QUEUE_SIZE);
+ shm_toc_insert(toc, num_toc++, mq);
+
+ mq_req_id = shm_toc_allocate(toc, sizeof(uint32));
+ shm_toc_insert(toc, num_toc++, mq_req_id);
+ *mq_req_id = 0;
+
+ qs_trace_slots = shm_toc_allocate(toc,
+
(Size) GPSC_TRACE_ID_LEN * (MaxBackends + 1));
+ shm_toc_insert(toc, num_toc++, qs_trace_slots);
+ memset(qs_trace_slots, 0, (Size) GPSC_TRACE_ID_LEN *
(MaxBackends + 1));
+ }
+ else
+ {
+ toc = shm_toc_attach(PG_QS_MODULE_KEY, shmem);
+ params = shm_toc_lookup(toc, num_toc++, false);
+ mq = shm_toc_lookup(toc, num_toc++, false);
+ mq_req_id = shm_toc_lookup(toc, num_toc++, false);
+ qs_trace_slots = shm_toc_lookup(toc, num_toc++, false);
+ }
+ LWLockRelease(AddinShmemInitLock);
+
+ if (prev_shmem_startup_hook)
+ prev_shmem_startup_hook();
+
+ module_initialized = true;
+}
+
+#if PG_VERSION_NUM >= 150000
+/*
+ * pg_qs_shmem_request -- hook called to request shared memory space.
+ *
+ * PostgreSQL 15+ separates the request phase from the startup phase.
+ * This hook is installed only when building against PG15+.
+ */
+static void
+pg_qs_shmem_request(void)
+{
+ if (prev_shmem_request_hook)
+ prev_shmem_request_hook();
+
+ RequestAddinShmemSpace(pg_qs_shmem_size());
+}
+#endif
+
+/*
+ * pg_qs_init -- initialise the pg_query_state signal infrastructure.
+ *
+ * Must be called from _PG_init() while
process_shared_preload_libraries_in_progress
+ * is true. Registers shared memory, custom ProcSignal handlers and GUC
+ * variables. Safe to call unconditionally for all roles.
+ */
+void
+pg_qs_init(void)
+{
+ if (!process_shared_preload_libraries_in_progress)
+ return;
+
+#if PG_VERSION_NUM >= 150000
+ prev_shmem_request_hook = shmem_request_hook;
+ shmem_request_hook = pg_qs_shmem_request;
+#else
+ RequestAddinShmemSpace(pg_qs_shmem_size());
+#endif
+
+ QueryStatePollReason = RegisterCustomProcSignalHandler(SendQueryState);
+ BackendInfoPollReason =
RegisterCustomProcSignalHandler(SendCdbComponents);
+
+ if (QueryStatePollReason == INVALID_PROCSIGNAL ||
+ BackendInfoPollReason == INVALID_PROCSIGNAL)
+ {
+ ereport(WARNING, (errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+ errmsg("pg_query_state isn't
loaded: insufficient custom ProcSignal slots")));
+ return;
+ }
+
+ DefineCustomBoolVariable("pg_query_state.enable",
+ "Enable module.",
+ NULL,
+ &pg_qs_enable,
+ true,
+ PGC_SUSET,
+ 0,
+ NULL, NULL, NULL);
+
+ DefineCustomBoolVariable("pg_query_state.enable_timing",
+ "Collect timing data,
not just row counts.",
+ NULL,
+ &pg_qs_timing,
+ true,
+ PGC_SUSET,
+ 0,
+ NULL, NULL, NULL);
+
+ DefineCustomBoolVariable("pg_query_state.enable_buffers",
+ "Collect buffer
usage.",
+ NULL,
+ &pg_qs_buffers,
+ true,
+ PGC_SUSET,
+ 0,
+ NULL, NULL, NULL);
+
+ prev_shmem_startup_hook = shmem_startup_hook;
+ shmem_startup_hook = pg_qs_shmem_startup;
+
+ /*
+ * Own the executor hooks rather than being called from the collector's
+ * wrappers: this module has to run outside of whatever else hooks the
+ * executor, because pg_qs_executor_start() only instruments a query
whose
+ * showstatctx is still unset, and gp_stats_collector allocates one
itself
+ * when gpsc.enable_analyze and gpsc.enable_cdbstats are on. A hook is
+ * pushed onto the head of the chain, so registering last means running
+ * first -- see _PG_init() in gp_stats_collector.c.
+ */
+ prev_ExecutorStart_hook = ExecutorStart_hook;
+ ExecutorStart_hook = pg_qs_executor_start;
+ prev_ExecutorRun_hook = ExecutorRun_hook;
+ ExecutorRun_hook = pg_qs_executor_run;
+ prev_ExecutorFinish_hook = ExecutorFinish_hook;
+ ExecutorFinish_hook = pg_qs_executor_finish;
+ prev_ExecutorEnd_hook = ExecutorEnd_hook;
+ ExecutorEnd_hook = pg_qs_executor_end;
+
+ elog(LOG, "pg_query_state: signal infrastructure initialised");
+}
+
+/* Executor lifecycle hooks */
+/*
+ * pg_qs_executor_start -- called at the start of executor execution.
+ *
+ * Enables instrumentation on the QueryDesc when:
+ * - pg_query_state is enabled
+ * - this is not an EXPLAIN-only execution
+ * - we are on a QD or QE role
+ * - there is no outer query already on the stack (top-level only)
+ * - the query passes the filter
+ * - no showstatctx is already attached
+ *
+ * Also assigns a synthetic queryId when the planner left it as zero.
+ *
+ * Parameters:
+ * queryDesc -- the QueryDesc being started
+ * eflags -- executor flags (EXEC_FLAG_EXPLAIN_ONLY etc.)
+ */
+static void
+pg_qs_executor_start(QueryDesc *queryDesc, int eflags)
+{
+ instr_time starttime;
+
+ if (pg_qs_enable
+ && ((eflags & EXEC_FLAG_EXPLAIN_ONLY) == 0)
+ && (Gp_role == GP_ROLE_DISPATCH || Gp_role == GP_ROLE_EXECUTE)
+ && is_querystack_empty()
+ && filter_query(queryDesc)
+ && queryDesc->showstatctx == NULL)
+ {
+ queryDesc->instrument_options |= INSTRUMENT_CDB;
+ queryDesc->instrument_options |= INSTRUMENT_ROWS;
+ if (pg_qs_timing)
+ queryDesc->instrument_options |= INSTRUMENT_TIMER;
+ if (pg_qs_buffers)
+ queryDesc->instrument_options |= INSTRUMENT_BUFFERS;
+
+ INSTR_TIME_SET_CURRENT(starttime);
+
+ /*
+ * cdbexplain_showExecStatsBegin() aggregates QE stats on the
QD and
+ * asserts Gp_role != GP_ROLE_EXECUTE, so it must run on the
dispatcher
+ * only. QE backends still get instrument_options above, which
is all
+ * the per-node walker reads.
+ */
+ if (Gp_role == GP_ROLE_DISPATCH)
+ queryDesc->showstatctx =
+ cdbexplain_showExecStatsBegin(queryDesc,
starttime);
+ queryDesc->totaltime = InstrAlloc(1, INSTRUMENT_ALL, false);
+
+ gpsc_reset_node_roll_state();
+ }
+
+ if (queryDesc->plannedstmt->queryId == 0)
+ queryDesc->plannedstmt->queryId =
+ ((uint64) gp_command_count << 32) + qs_push_count;
+
+ if (prev_ExecutorStart_hook)
+ prev_ExecutorStart_hook(queryDesc, eflags);
+ else
+ standard_ExecutorStart(queryDesc, eflags);
+}
+
+/*
+ * pg_qs_executor_run -- called when the executor begins fetching tuples.
+ *
+ * Keeps the QueryDesc on the stack for as long as tuples are being fetched, so
+ * that a poll arriving mid-run finds it.
+ */
+static void
+pg_qs_executor_run(QueryDesc *queryDesc, ScanDirection direction,
+ uint64 count, bool execute_once)
+{
+ push_query(queryDesc);
+ PG_TRY();
+ {
+ if (prev_ExecutorRun_hook)
+ prev_ExecutorRun_hook(queryDesc, direction, count,
execute_once);
+ else
+ standard_ExecutorRun(queryDesc, direction, count,
execute_once);
+ }
+ PG_FINALLY();
+ {
+ pg_qs_pop_query();
+ }
+ PG_END_TRY();
+}
+
+/*
+ * pg_qs_executor_finish -- called after all tuples have been fetched.
+ *
+ * Same push/pop as the run phase: the query stays visible to signal handlers
+ * while after-triggers and the like are still running.
+ */
+static void
+pg_qs_executor_finish(QueryDesc *queryDesc)
+{
+ push_query(queryDesc);
+ PG_TRY();
+ {
+ if (prev_ExecutorFinish_hook)
+ prev_ExecutorFinish_hook(queryDesc);
+ else
+ standard_ExecutorFinish(queryDesc);
+ }
+ PG_FINALLY();
+ {
+ pg_qs_pop_query();
+ }
+ PG_END_TRY();
+}
+
+/*
+ * pg_qs_executor_end -- called when executor resources are released.
+ *
+ * Drops the per-node rolling state so the next query on this backend starts
its
+ * delta accounting clean. It does not collect or push anything: a finish is
not
+ * a signalled collection and carries no trace_id to key a batch under.
+ */
+static void
+pg_qs_executor_end(QueryDesc *queryDesc)
+{
+ if (queryDesc && pg_qs_enable)
+ gpsc_reset_node_roll_state();
+
+ if (prev_ExecutorEnd_hook)
+ prev_ExecutorEnd_hook(queryDesc);
+ else
+ standard_ExecutorEnd(queryDesc);
+}
+
+static void
+push_query(QueryDesc *queryDesc)
+{
+ qs_push_count++;
+ QueryDescStack = lcons(queryDesc, QueryDescStack);
+}
+
+static void
+pg_qs_pop_query(void)
+{
+ QueryDescStack = list_delete_first(QueryDescStack);
+}
+
+static bool
+is_querystack_empty(void)
+{
+ return list_length(QueryDescStack) == 0;
+}
+
+QueryDesc *
+get_toppest_query(void)
+{
+ return (QueryDescStack == NIL) ? NULL : (QueryDesc *)
llast(QueryDescStack);
+}
+
+/*
+ * filter_query -- decide whether to instrument a given QueryDesc.
+ *
+ * Returns false for cursor queries with non-default cursor options, and for
+ * utility statements. Returns true for SELECT, INSERT, UPDATE, DELETE, MERGE.
+ */
+static bool
+filter_query(QueryDesc *queryDesc)
+{
+ Portal portal;
+
+ if (queryDesc == NULL)
+ return false;
+
+ if (queryDesc->extended_query && queryDesc->portal_name)
+ {
+ portal = GetPortalByName(queryDesc->portal_name);
+ if (!PointerIsValid(portal) || portal->cursorOptions !=
CURSOR_OPT_NO_SCROLL)
+ return false;
+ }
+
+ return (queryDesc->operation == CMD_SELECT ||
+ queryDesc->operation == CMD_DELETE ||
+ queryDesc->operation == CMD_INSERT ||
+ queryDesc->operation == CMD_UPDATE ||
+ queryDesc->operation == CMD_MERGE);
+}
+
+/*
+ * LockShmem -- acquire an exclusive user-lock keyed by (PG_QS_MODULE_KEY,
key).
+ *
+ * Used to serialise access to the shared mq between concurrent requestors
+ * and between requestor and handler.
+ */
+static void
+LockShmem(LOCKTAG *tag, uint32 key)
+{
+ LockAcquireResult result;
+
+ tag->locktag_field1 = PG_QS_MODULE_KEY;
+ tag->locktag_field2 = key;
+ tag->locktag_field3 = 0;
+ tag->locktag_field4 = 0;
+ tag->locktag_type = LOCKTAG_USERLOCK;
+ tag->locktag_lockmethodid = USER_LOCKMETHOD;
+
+ result = LockAcquire(tag, ExclusiveLock, false, false);
+ Assert(result == LOCKACQUIRE_OK);
+}
+
+/*
+ * UnlockShmem -- release the exclusive user-lock acquired by LockShmem.
+ */
+static void
+UnlockShmem(LOCKTAG *tag)
+{
+ LockRelease(tag, ExclusiveLock, false);
+}
+
+/*
+ * GetRemoteBackendInfo -- obtain the list of (segid, pid) pairs from QD.
+ *
+ * Sends BackendInfoPollReason to proc and waits for the reply. On success,
+ * *result is populated with gp_segment_pid entries (palloc'd).
+ *
+ * Returns the PG_QS_RequestResult code from the reply.
+ */
+static PG_QS_RequestResult
+GetRemoteBackendInfo(PGPROC *proc, List **result)
+{
+ int sig_result;
+ shm_mq_handle *mqh;
+ shm_mq_result mq_receive_result;
+ Size msg_len;
+ backend_info *msg;
+ LOCKTAG tag;
+ int i;
+
+ LockShmem(&tag, PG_QS_SND_KEY);
+ params->reason = BackendInfoPollReason;
+ mq = shm_mq_create(mq, QUEUE_SIZE);
+ shm_mq_set_sender(mq, proc);
+ shm_mq_set_receiver(mq, MyProc);
+ *mq_req_id = reqid;
+ UnlockShmem(&tag);
+
+ sig_result = SendProcSignal(proc->pid, BackendInfoPollReason,
+
proc->backendId);
+ if (sig_result == -1)
+ ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("could not send
BackendInfoPollReason signal")));
+
+ mqh = shm_mq_attach(mq, NULL, NULL);
+ mq_receive_result = shm_mq_receive_with_timeout(mqh, &msg_len,
+
(void **) &msg,
+
MAX_RCV_TIMEOUT);
+
+ if (mq_receive_result != SHM_MQ_SUCCESS || msg == NULL ||
+ msg->reqid != (uint32) reqid)
+ {
+ shm_mq_detach(mqh);
+ ereport(WARNING, (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("GetRemoteBackendInfo:
message not received")));
+ return QUERY_NOT_RUNNING;
+ }
+
+ if (msg->result_code != QS_RETURNED)
+ {
+ PG_QS_RequestResult result_code = msg->result_code;
+ shm_mq_detach(mqh);
+ return result_code;
+ }
+
+ /* Validate the reply payload length against the reported backend
count. */
+ {
+ int expected_len = BASE_SIZEOF_GP_BACKEND_INFO +
+ msg->number *
sizeof(gp_segment_pid);
+ if ((int) msg_len != expected_len)
+ {
+ shm_mq_detach(mqh);
+ ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
+
errmsg("GetRemoteBackendInfo: unexpected message length")));
+ }
+ }
+
+ for (i = 0; i < msg->number; i++)
+ {
+ gp_segment_pid *segpid = palloc(sizeof(gp_segment_pid));
+ *segpid = msg->pids[i];
+ *result = lcons(segpid, *result);
+ }
+
+ shm_mq_detach(mqh);
+ return QS_RETURNED;
+}
+
+/*
+ * CollectQEQueryState -- fan-out query-state signals to the segment QEs.
+ *
+ * Dispatches a cbdb_mpp_query_state() call to each segment listed in
+ * backendInfo. Results are returned as raw CdbPgResults.
+ *
+ * GPSC_SEGID_ENTRY_DB entries are left out: the dispatch reaches primary
+ * segments only, and there the receiving cbdb_mpp_query_state() matches
+ * entries against its own GpIdentity.segindex, which is never negative.
+ * SignalEntryDbBackends() handles those.
+ */
+static void
+CollectQEQueryState(List *backendInfo, bytea *trace_id)
+{
+ ListCell *lc;
+ StringInfoData params_buf;
+ char *sql;
+ char trace_id_hex[2 * GPSC_TRACE_ID_LEN + 1];
+ int nsegments = 0;
+
+ if (list_length(backendInfo) == 0)
+ return;
+
+ initStringInfo(¶ms_buf);
+
+ foreach(lc, backendInfo)
+ {
+ gp_segment_pid *segpid = (gp_segment_pid *) lfirst(lc);
+
+ if (segpid->segid < 0)
+ continue;
+
+ if (nsegments++ > 0)
+ appendStringInfoChar(¶ms_buf, ',');
+ appendStringInfo(¶ms_buf, "'(%d,%d)'", segpid->segid,
segpid->pid);
+ }
+
+ if (nsegments == 0)
+ {
+ pfree(params_buf.data);
+ return;
+ }
+
+ hex_encode(VARDATA_ANY(trace_id), GPSC_TRACE_ID_LEN, trace_id_hex);
+ trace_id_hex[2 * GPSC_TRACE_ID_LEN] = '\0';
+ sql = psprintf("SELECT
gpsc.cbdb_mpp_query_state((ARRAY[%s])::gpsc.gp_segment_pid[], '\\x%s'::bytea)",
+ params_buf.data, trace_id_hex);
+
+ CdbDispatchCommand(sql, DF_NONE, NULL);
+ pfree(params_buf.data);
+ pfree(sql);
+}
+
+/*
+ * SignalEntryDbBackends -- poll the entry-db QEs listed in backendInfo.
+ *
+ * An entry-db reader runs the coordinator-side slice of a distributed query
and
+ * so holds the only instrumentation for it, but it lives in the coordinator's
+ * own postmaster and no dispatch reaches it. Since it is a local backend, the
+ * QD signals it the same way it signals itself.
+ */
+static void
+SignalEntryDbBackends(List *backendInfo, bytea *trace_id)
+{
+ ListCell *lc;
+
+ foreach(lc, backendInfo)
+ {
+ gp_segment_pid *segpid = (gp_segment_pid *) lfirst(lc);
+ PGPROC *proc;
+
+ if (segpid->segid != GPSC_SEGID_ENTRY_DB)
+ continue;
+
+ proc = BackendPidGetProc(segpid->pid);
+ if (!proc || proc->backendId == InvalidBackendId)
+ continue;
+
+ memcpy(qs_trace_slots[proc->backendId], VARDATA_ANY(trace_id),
+ GPSC_TRACE_ID_LEN);
+ if (SendProcSignal(proc->pid, QueryStatePollReason,
+ proc->backendId) == -1)
+ elog(DEBUG1, "pg_query_state: failed to signal entry-db
backend pid=%d",
+ segpid->pid);
+ }
+}
+
+/*
+ * shm_mq_receive_with_timeout -- receive from mqh, blocking up to `timeout`
ms.
+ *
+ * Calls receive_msg_by_parts() in a loop, sleeping on the latch between
+ * retries. Returns SHM_MQ_SUCCESS, SHM_MQ_DETACHED, or SHM_MQ_WOULD_BLOCK
+ * (the last meaning the timeout expired).
+ *
+ * On success, *nbytesp is set to the message length and *datap to a palloc'd
+ * buffer containing the message.
+ */
+static shm_mq_result
+shm_mq_receive_with_timeout(shm_mq_handle *mqh,
+ Size *nbytesp,
+ void **datap,
+ int64 timeout)
+{
+ int rc = 0;
+ int64 delay = timeout;
+ instr_time start_time;
+ instr_time cur_time;
+
+ INSTR_TIME_SET_CURRENT(start_time);
+
+ for (;;)
+ {
+ shm_mq_result result;
+
+ result = receive_msg_by_parts(mqh, nbytesp, datap, timeout,
&rc, true);
+ if (result != SHM_MQ_WOULD_BLOCK)
+ return result;
+
+ if (rc & WL_TIMEOUT || delay <= 0)
+ return SHM_MQ_WOULD_BLOCK;
+
+ rc = WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_EXIT_ON_PM_DEATH |
WL_TIMEOUT,
+ delay, PG_WAIT_EXTENSION);
+
+ INSTR_TIME_SET_CURRENT(cur_time);
+ INSTR_TIME_SUBTRACT(cur_time, start_time);
+ delay = timeout - (int64) INSTR_TIME_GET_MILLISEC(cur_time);
+ if (delay <= 0)
+ return SHM_MQ_WOULD_BLOCK;
+
+ CHECK_FOR_INTERRUPTS();
+ ResetLatch(MyLatch);
+ }
+}
+
+/*
+ * receive_msg_by_parts -- reassemble a multi-chunk message from mqh.
+ *
+ * The wire protocol prefixes each message with its total byte count (a Size),
+ * followed by one or more chunks of up to MSG_MAX_SIZE bytes. This function
+ * reads the prefix, allocates a buffer, and loops until all chunks arrive.
+ *
+ * Parameters:
+ * mqh -- attached message-queue handle
+ * total -- out: total bytes received
+ * datap -- out: palloc'd buffer with reassembled message
+ * timeout -- caller's deadline in ms (used only for PART_RCV_DELAY retries)
+ * rc -- out: WaitLatch flags (set to WL_TIMEOUT if we give up)
+ * nowait -- passed through to shm_mq_receive
+ */
+static shm_mq_result
+receive_msg_by_parts(shm_mq_handle *mqh, Size *total, void **datap,
+ int64 timeout, int *rc, bool nowait)
+{
+ shm_mq_result mq_receive_result;
+ shm_mq_msg *buff;
+ int offset;
+ Size *expected;
+ Size expected_data;
+ Size len;
+
+ /* Read the length prefix. */
+ mq_receive_result = shm_mq_receive(mqh, &len, (void **) &expected,
nowait);
+ if (mq_receive_result != SHM_MQ_SUCCESS)
+ return mq_receive_result;
+ Assert(len == sizeof(Size));
+
+ expected_data = *expected;
+ Assert(expected_data < UINT32_MAX);
+ *datap = palloc0(expected_data);
+
+ /* Reassemble chunks until we have expected_data bytes. */
+ for (offset = 0; offset < (int) expected_data; )
+ {
+ int64 delay = timeout;
+
+ for (;;)
+ {
+ mq_receive_result = shm_mq_receive(mqh, &len, (void **)
&buff,
+
nowait);
+ if (mq_receive_result != SHM_MQ_SUCCESS)
+ {
+ if (nowait && mq_receive_result ==
SHM_MQ_WOULD_BLOCK)
+ {
+ if (delay > 0)
+ {
+ pg_usleep(PART_RCV_DELAY *
1000);
+ delay -= PART_RCV_DELAY;
+ continue;
+ }
+ if (rc)
+ *rc |= WL_TIMEOUT;
+ }
+ return mq_receive_result;
+ }
+ break;
+ }
+ memcpy((char *) *datap + offset, buff, len);
+ offset += len;
+ }
+
+ *total = offset;
+ return mq_receive_result;
+}
+
+/*
+ * qs_fetch_backend_info -- serialise a backend-info request and collect the
+ * (segid, pid) list for the query running on `proc`.
+ *
+ * Holds PG_QS_RCV_KEY across the request so concurrent requestors do not
clobber
+ * the shared mq, releasing it on both the success and error paths.
+ */
+static PG_QS_RequestResult
+qs_fetch_backend_info(PGPROC *proc, List **backend_info)
+{
+ LOCKTAG tag;
+ PG_QS_RequestResult result;
+
+ LockShmem(&tag, PG_QS_RCV_KEY);
+ PG_TRY();
+ {
+ reqid = *mq_req_id + 1;
+ result = GetRemoteBackendInfo(proc, backend_info);
+ UnlockShmem(&tag);
+ }
+ PG_CATCH();
+ {
+ UnlockShmem(&tag);
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+
+ return result;
+}
+
+/* SQL callable functions */
+/*
+ * pg_query_state -- entry point for the pg_query_state() SQL function.
+ *
+ * Obtains the user-id and segment-backend list from the target backend,
+ * then fans out cbdb_mpp_query_state() to each QE.
+ */
+PG_FUNCTION_INFO_V1(pg_query_state);
+Datum
+pg_query_state(PG_FUNCTION_ARGS)
+{
+ pid_t pid = PG_GETARG_INT32(0);
+ bytea *trace_id = PG_GETARG_BYTEA_P(1);
+ PGPROC *proc;
+ PG_QS_RequestResult result;
+ List *backend_info = NIL;
+
+ if (VARSIZE_ANY_EXHDR(trace_id) != GPSC_TRACE_ID_LEN)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid size of trace_id: %zu,
expected %d",
+ VARSIZE_ANY_EXHDR(trace_id),
GPSC_TRACE_ID_LEN)));
+
+ if (pid == MyProcPid)
+ ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot extract state of
current process")));
+
+ if (!module_initialized)
+ ereport(ERROR,
+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("pg_query_state must be loaded via
shared_preload_libraries")));
+
+ proc = BackendPidGetProc(pid);
+ if (!proc || proc->backendId == InvalidBackendId ||
+ proc->databaseId == InvalidOid || proc->roleId == InvalidOid)
+ ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("backend with pid=%d not
found", pid)));
+
+ if (!(superuser() || GetUserId() == proc->roleId))
+ {
+ ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied")));
+ }
+
+ result = qs_fetch_backend_info(proc, &backend_info);
+
+ switch (result)
+ {
+ case QUERY_NOT_RUNNING:
+ elog(DEBUG1, "pg_query_state: pid=%d is not running a
query", pid);
+ break;
+
+ case STAT_DISABLED:
+ elog(DEBUG1, "pg_query_state: stats collection
disabled");
+ break;
+
+ case WRONG_ROLE:
+ /*
+ * Not the QD, so there is no participant list to fan
out to and no
+ * point signalling: a QE polled directly would report
a single
+ * slice that no collection is waiting for. Stay quiet
here -- the
+ * caller-facing complaint belongs to
pg_query_state_backends(),
+ * which errors out on the same result code.
+ */
+ elog(DEBUG1, "pg_query_state: pid=%d is a query
executor, not the QD", pid);
+ break;
+
+ case QS_RETURNED:
+ /*
+ * Signal all segment QEs to push their plan-node stats
via UDS,
+ * carrying the trace_id so every backend's batch lands
under the one
+ * key this pg_query_state() invocation owns.
+ */
+ CollectQEQueryState(backend_info, trace_id);
+ SignalEntryDbBackends(backend_info, trace_id);
+
+ /*
+ * Signal the QD backend itself so it pushes
coordinator-side plan
+ * nodes and the plan-doc. SendQueryState() emits
directly via UDS.
+ * Stamp the target's own trace slot before signalling,
so its batch
+ * lands under this collection's key.
+ */
+ memcpy(qs_trace_slots[proc->backendId],
VARDATA_ANY(trace_id),
+ GPSC_TRACE_ID_LEN);
+ SendProcSignal(proc->pid, QueryStatePollReason,
proc->backendId);
+ break;
+ }
+
+ PG_RETURN_VOID();
+}
+
+/*
+ * pg_query_state_backends -- list the QE backends participating in the query
+ * running on backend `pid`.
+ *
+ * Returns a set of (segid, pid) rows obtained from the coordinator via
+ * GetRemoteBackendInfo (the same list the poll path fans out to). A consumer
+ * can use the row count as the expected number of backends that will report.
+ *
+ * Uses the materialize SRF mode: the whole list is built into a tuplestore in
+ * one call. Returns an empty set when the target query is not running.
+ */
+PG_FUNCTION_INFO_V1(pg_query_state_backends);
+Datum
+pg_query_state_backends(PG_FUNCTION_ARGS)
+{
+ pid_t pid = PG_GETARG_INT32(0);
+ ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
+ TupleDesc tupdesc;
+ Tuplestorestate *tupstore;
+ PGPROC *proc;
+ List *backend_info = NIL;
+ PG_QS_RequestResult result;
+ ListCell *lc;
+
+ InitMaterializedSRF(fcinfo, 0);
+ tupdesc = rsinfo->setDesc;
+ tupstore = rsinfo->setResult;
+
+ if (pid == MyProcPid)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot extract state of current
process")));
+
+ proc = BackendPidGetProc(pid);
+ if (!proc || proc->backendId == InvalidBackendId ||
+ proc->databaseId == InvalidOid || proc->roleId == InvalidOid)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("backend with pid=%d not found", pid)));
+
+ if (!module_initialized)
+ {
+ ereport(ERROR,
+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("pg_query_state must be loaded via
shared_preload_libraries")));
+ }
+
+ if (!(superuser() || GetUserId() == proc->roleId))
+ {
+ ereport(ERROR, (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied")));
+ }
+
+ result = qs_fetch_backend_info(proc, &backend_info);
+
+ /*
+ * Not running / disabled are ordinary outcomes of polling a pid that
has
+ * just finished: return an empty set rather than erroring. A
wrong-role
+ * target is different -- the backend is alive and will never answer,
which
+ * a caller must be able to tell apart from a finished query, so that
one
+ * does error out.
+ */
+ switch (result)
+ {
+ case QUERY_NOT_RUNNING:
+ elog(DEBUG1, "pg_query_state_backends: pid=%d is not
running a query", pid);
+ return (Datum) 0;
+
+ case STAT_DISABLED:
+ elog(DEBUG1, "pg_query_state_backends: stats collection
disabled");
+ return (Datum) 0;
+
+ case WRONG_ROLE:
+ ereport(ERROR,
+
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("backend with pid=%d is a query
executor, "
+ "not the session's
coordinator backend", pid)));
+ break;
+
+ case QS_RETURNED:
+ break;
+ }
+
+ foreach(lc, backend_info)
+ {
+ gp_segment_pid *segpid = (gp_segment_pid *) lfirst(lc);
+ Datum values[2];
+ bool nulls[2] = {false, false};
+
+ values[0] = Int32GetDatum(segpid->segid);
+ values[1] = Int32GetDatum(segpid->pid);
+ tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+ }
+
+ /*
+ * QD-only query (INSERT ... VALUES, catalog reads, and other
coordinator-
+ * local plans): no QE gang ran, so backend_info is empty even though
the
+ * coordinator is executing and will push its own per-node batch.
Report the
+ * coordinator itself so the caller does not mistake an empty QE list
for a
+ * finished query and drop the QD's batch.
+ */
+ if (list_length(backend_info) == 0)
+ {
+ Datum values[2];
+ bool nulls[2] = {false, false};
+
+ values[0] = Int32GetDatum(GPSC_SEGID_QD);
+ values[1] = Int32GetDatum(proc->pid);
+ tuplestore_putvalues(tupstore, tupdesc, values, nulls);
+ }
+
+ return (Datum) 0;
+}
+
+/*
+ * cbdb_mpp_query_state -- QE-side entry point dispatched by
CollectQEQueryState.
+ *
+ * Receives an array of gp_segment_pid, filters those belonging to this
+ * segment, and fires QueryStatePollReason at each matching backend.
+ */
+PG_FUNCTION_INFO_V1(cbdb_mpp_query_state);
+Datum
+cbdb_mpp_query_state(PG_FUNCTION_ARGS)
+{
+ ListCell *iter;
+ List *alive_procs =
get_query_backend_info(PG_GETARG_ARRAYTYPE_P(0));
+ bytea *trace_id = PG_GETARG_BYTEA_P(1);
+
+ if (VARSIZE_ANY_EXHDR(trace_id) != GPSC_TRACE_ID_LEN)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid size of trace_id: %zu,
expected %d",
+ VARSIZE_ANY_EXHDR(trace_id),
GPSC_TRACE_ID_LEN)));
+
+ if (alive_procs == NIL)
+ PG_RETURN_NULL();
+
+ foreach(iter, alive_procs)
+ {
+ PGPROC *proc = (PGPROC *) lfirst(iter);
+ int sig_result;
+
+ if (!proc || proc->backendId == InvalidBackendId)
+ continue;
+
+ /* Stamp the target's own trace slot before signalling it. */
+ memcpy(qs_trace_slots[proc->backendId], VARDATA_ANY(trace_id),
+ GPSC_TRACE_ID_LEN);
+
+ sig_result = SendProcSignal(proc->pid, QueryStatePollReason,
+
proc->backendId);
Review Comment:
The internal QE dispatch function is granted to `PUBLIC`, but
`cbdb_mpp_query_state()` performs no ownership or superuser check before
copying caller-supplied `(segid,pid)` values into trace slots and sending
`QueryStatePollReason`. A user who can execute this function on a QE can
therefore trigger collection of another role's live plan and push it to UDS,
bypassing the gates on the public coordinator APIs. Keep this entry point
non-public or enforce the same target-backend authorization for every signalled
process.
##########
gpcontrib/gp_stats_collector/src/PlanNodeEmitter.cpp:
##########
@@ -0,0 +1,170 @@
+/*-------------------------------------------------------------------------
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ * PlanNodeEmitter.cpp
+ * Build and send per-node protobuf messages to the yagpcc UDS
sink.
+ *
+ * This file is the bridge between the C pg_query_state layer and the C++
+ * protobuf / UDS connector infrastructure. It implements the functions
+ * declared in PlanNodeEmitter.h and callable from plain C:
+ *
+ * gpsc_qs_sync_config() -- reload the Config singleton
+ * gpsc_emit_node_batch() -- serialize a plan-tree snapshot and send it
+ * gpsc_emit_query_plan() -- serialize a plan document and send it
+ *
+ * The outgoing message types are yagpcc::SetPerNodeBatchReq and
+ * yagpcc::SetQueryPlanReq (generated from protos/yagpcc_set_per_node.proto).
+ * Transmission is handled by UDSConnector, which prepends the 8-byte extended
+ * protocol header before writing to the socket.
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_stats_collector/src/PlanNodeEmitter.cpp
+ */
+
+#include "PlanNodeEmitter.h"
+#include "protos/yagpcc_set_per_node.pb.h"
+#include "UDSConnector.h"
+#include "Config.h"
+#include "ProtoUtils.h"
+
+/* Module-private Config instance shared across all emit calls in a session. */
+static Config pne_config;
+
+/*
+ * gpsc_qs_sync_config -- reload the Config singleton.
+ *
+ * Must be called before a gpsc_emit_node_batch() call so that the UDS path
+ * and other settings are up to date. It is a no-op when the config has not
+ * changed since the last call.
+ */
+extern "C" void
+gpsc_qs_sync_config()
+{
+ pne_config.sync();
+}
+
+/*
+ * map_node_status -- convert a QsNodeStatus enum to yagpcc::PlanNodeStatus.
+ *
+ * Returns PLAN_NODE_STATUS_UNSPECIFIED for any value not recognised by the
+ * switch, which is safe because the receiver ignores unknown status codes.
+ */
+static yagpcc::PlanNodeStatus
+map_node_status(QsNodeStatus status)
+{
+ switch (status)
+ {
+ case QS_NODE_STATUS_INITIALIZED:
+ return yagpcc::PLAN_NODE_STATUS_INITIALIZED;
+ case QS_NODE_STATUS_EXECUTING:
+ return yagpcc::PLAN_NODE_STATUS_EXECUTING;
+ case QS_NODE_STATUS_FINISHED:
+ return yagpcc::PLAN_NODE_STATUS_FINISHED;
+ default:
+ return yagpcc::PLAN_NODE_STATUS_UNSPECIFIED;
+ }
+}
+
+extern "C" void
+gpsc_emit_node_batch(GpscNodeSample **nodes, int count, const char *trace_id)
+{
+ if (count <= 0)
+ return;
Review Comment:
This C bridge is invoked directly from the signal handler, but it has no C++
exception boundary. `Config::sync()`, protobuf construction/serialization, and
`gpdb::palloc()` can throw; unlike the existing `cpp_call()` wrapper in
`hook_wrappers.cpp`, an exception escaping this `extern "C"` path is not
translated into a PostgreSQL error and can terminate the backend. Wrap all
three emitter entry points in the established exception-handling adapter (or
catch and report locally).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
