Add wait stats

gpcontrib/gp_stats_collector/metric.md

@@ -69,6 +69,7 @@ submit -> ExecutorStart() -> start -> ExecutorRun() -> ExecutorFinish() -> end -
 | `blk_write_time`             | double | E, D    | ABS      | +      | Node    |     +      |    +    | seconds | Time writing data blocks                            |
 | `inherited_calls`            | uint64 | E, D    | ABS      | -      | Node    |     +      |    +    | count   | Nested query count (GPSC-specific)                |
 | `inherited_time`             | double | E, D    | ABS      | -      | Node    |     +      |    +    | seconds | Nested query time (GPSC-specific)                 |
+| `cross_slice_wait_ms`        | double | E, D    | ABS      | +      | Node    |     +      |    +    | ms**    | Longest cross-slice ShareInputScan wait (GPSC-specific) |
 | **NetworkStat (sent)**       |        |         |          |        |         |            |         |         |                                                     |
 | `sent.total_bytes`           | uint32 | D       | ABS      | -      | Node    |     +      |    +    | bytes   | Bytes sent, including headers                       |
 | `sent.tuple_bytes`           | uint32 | D       | ABS      | -      | Node    |     +      |    +    | bytes   | Bytes of pure tuple-data sent                       |
@@ -122,5 +123,18 @@ submit -> ExecutorStart() -> start -> ExecutorRun() -> ExecutorFinish() -> end -
 | `dbid`                       | int32  | All     | ABS      | -      | Node    |     +      |    +    | id      | Database ID                                         |
 | `segment_index`              | int32  | All     | ABS      | -      | Node    |     +      |    +    | id      | Segment index (-1=coordinator)                      |
 
+**\*\* `cross_slice_wait_ms`** is how long a cross-slice ShareInputScan consumer
+blocked waiting for its producer slice to publish the shared tuplestore. Read as
+max across segments against the average: a wide gap means the producer finished
+unevenly, not that the consumer is slow. It also rises when the producer slice is
+starved of CPU or interconnect, and it is the only trace left by a query killed on
+`statement_timeout` while blocked there.
+
+Two things set it apart from the rest of the table. It is reported in milliseconds
+rather than seconds -- the unit is part of the field name, so the wire contract
+stays unambiguous. And it is the one `MetricInstrumentation` field filled without
+instrumentation enabled: the executor measures the wait either way. A zero wait is
+not reported.
+
 ---
 

gpcontrib/gp_stats_collector/protos/gpsc_metrics.proto

@@ -175,6 +175,28 @@ message MetricInstrumentation {
     uint64  inherited_calls    = 22; /* the number of executed sub-queries */
     double  inherited_time     = 23; /* total time spend on inherited execution */
     InterconnectStat interconnect = 24;
+    /*
+     * Longest time a cross-slice ShareInputScan consumer spent blocked waiting
+     * for its producer slice to publish the shared tuplestore. Scope is the
+     * reporting process (one per slice), not the segment as a whole. Unlike
+     * the other timings here it is in milliseconds, hence the suffix.
+     *
+     * The wait says how long one slice sat idle on another, which is worth
+     * watching well beyond a mis-wired share:
+     *   - aggregated as max across segments and compared against the average,
+     *     it measures how unevenly the producer finished -- large max/avg gap
+     *     means a skewed producer, not a slow consumer;
+     *   - it splits "this node is slow" from "this node is waiting", so time
+     *     is not attributed to the consumer that merely blocked;
+     *   - it grows when the producer slice is starved of CPU or interconnect,
+     *     making it an early signal of cluster-wide contention;
+     *   - it is the only trace left by a query killed on statement_timeout
+     *     while blocked here, where per-node instrumentation never arrives.
+     *
+     * Kept in sync by hand with protos/yagpcc_metrics.proto in the
+     * yagp_hooks_collector repo, which this directory forked from.
+     */
+    double  cross_slice_wait_ms = 25;
 }
 
 message SpillInfo {

gpcontrib/gp_stats_collector/src/ProtoUtils.cpp

@@ -250,6 +250,23 @@ set_gp_metrics(gpsc::GPMetrics *metrics, QueryDesc *query_desc,
 		set_metric_instrumentation(metrics->mutable_instrumentation(),
 								   query_desc, nested_calls, nested_time);
 	}
+
+	/*
+	 * Longest cross-slice ShareInputScan wait seen by this process, rolled up
+	 * to the query level in the executor. Reported in milliseconds; yagpcc
+	 * takes the max across segments, so a large gap against the average marks
+	 * a skewed producer slice rather than a slow consumer.
+	 *
+	 * Set outside the block above on purpose: the executor measures the wait
+	 * regardless of instrumentation, and a query killed by statement_timeout --
+	 * the case with no other trace left -- may well run without it. Only a
+	 * non-zero wait is reported, so queries that never waited do not get an
+	 * otherwise empty instrumentation submessage.
+	 */
+	if (query_desc->estate && query_desc->estate->es_cross_slice_wait > 0)
+		metrics->mutable_instrumentation()->set_cross_slice_wait_ms(
+			query_desc->estate->es_cross_slice_wait * 1000.0);
+
 	fill_self_stats(metrics->mutable_systemstat());
 	metrics->mutable_systemstat()->set_runningtimeseconds(
 		time(NULL) - metrics->mutable_systemstat()->runningtimeseconds());

gpcontrib/gp_stats_collector/src/log/LogSchema.h

@@ -112,6 +112,7 @@ inline constexpr std::array log_tbl_desc = {
  LogDesc{"instrumentation_blk_write_time",    "query_metrics.instrumentation.blk_write_time", FLOAT8OID},  
  LogDesc{"instrumentation_startup_time",      "query_metrics.instrumentation.startup_time",   FLOAT8OID},  
  LogDesc{"instrumentation_inherited_time",    "query_metrics.instrumentation.inherited_time", FLOAT8OID},  
+ LogDesc{"instrumentation_cross_slice_wait_ms", "query_metrics.instrumentation.cross_slice_wait_ms", FLOAT8OID},  
  LogDesc{"datetime",            "datetime",                       TIMESTAMPTZOID},  
  LogDesc{"submit_time",         "submit_time",                    TIMESTAMPTZOID},  
  LogDesc{"start_time",          "start_time",                     TIMESTAMPTZOID},  

src/backend/commands/explain_gp.c

@@ -112,6 +112,8 @@ typedef struct CdbExplain_StatInst
 	ExplainSortMethod sortMethod;	/* Type of sort */
 	ExplainSortSpaceType sortSpaceType; /* Sort space type */
 	long		sortSpaceUsed;	/* Memory / Disk used by sort(KBytes) */
+	double		shareinputwait; /* ShareInputScan: seconds blocked on another
+								 * slice */
 	int			bnotes;			/* Offset to beginning of node's extra text */
 	int			enotes;			/* Offset to end of node's extra text */
 } CdbExplain_StatInst;
@@ -182,6 +184,14 @@ typedef struct CdbExplain_NodeSummary
 	/* Summary of space used by sort */
 	CdbExplain_Agg sortSpaceUsed[NUM_SORT_SPACE_TYPE][NUM_SORT_METHOD];
 
+	/*
+	 * Time a cross-slice ShareInputScan spent blocked on another slice.
+	 * Aggregated across workers, because the per-worker extra message text is
+	 * selected by memory/row-count criteria and would otherwise report an
+	 * arbitrary worker's wait rather than the worst one.
+	 */
+	CdbExplain_Agg shareinputwait;
+
 	/* insts array info */
 	int			segindex0;		/* segment id of insts[0] */
 	int			ninst;			/* num of StatInst entries in inst array */
@@ -995,6 +1005,14 @@ cdbexplain_collectStatsFromNode(PlanState *planstate, CdbExplain_SendStatCtx *ct
 	si->sortMethod = String2ExplainSortMethod(instr->sortMethod);
 	si->sortSpaceType = String2ExplainSortSpaceType(instr->sortSpaceType, si->sortMethod);
 	si->sortSpaceUsed = instr->sortSpaceUsed;
+
+	if (IsA(planstate, ShareInputScanState))
+	{
+		ShareInputScanState *sisstate = (ShareInputScanState *) planstate;
+
+		si->shareinputwait =
+			INSTR_TIME_GET_DOUBLE(sisstate->waitready_time);
+	}
 }								/* cdbexplain_collectStatsFromNode */
 
 
@@ -1121,6 +1139,7 @@ cdbexplain_depositStatsToNode(PlanState *planstate, CdbExplain_RecvStatCtx *ctx)
 	CdbExplain_DepStatAcc memory_accounting_global_peak;
 	CdbExplain_DepStatAcc peakMemBalance;
 	CdbExplain_DepStatAcc totalPartTableScanned;
+	CdbExplain_DepStatAcc shareinputwait;
 	CdbExplain_DepStatAcc sortSpaceUsed[NUM_SORT_SPACE_TYPE][NUM_SORT_METHOD];
 	int			imsgptr;
 	int			nInst;
@@ -1146,6 +1165,7 @@ cdbexplain_depositStatsToNode(PlanState *planstate, CdbExplain_RecvStatCtx *ctx)
 	cdbexplain_depStatAcc_init0(&totalWorkfileCreated);
 	cdbexplain_depStatAcc_init0(&peakMemBalance);
 	cdbexplain_depStatAcc_init0(&totalPartTableScanned);
+	cdbexplain_depStatAcc_init0(&shareinputwait);
 	for (int idx = 0; idx < NUM_SORT_METHOD; ++idx)
 	{
 		cdbexplain_depStatAcc_init0(&sortSpaceUsed[MEMORY_SORT_SPACE_TYPE - 1][idx]);
@@ -1190,6 +1210,7 @@ cdbexplain_depositStatsToNode(PlanState *planstate, CdbExplain_RecvStatCtx *ctx)
 		cdbexplain_depStatAcc_upd(&totalWorkfileCreated, (rsi->workfileCreated ? 1 : 0), rsh, rsi, nsi);
 		cdbexplain_depStatAcc_upd(&peakMemBalance, rsi->peakMemBalance, rsh, rsi, nsi);
 		cdbexplain_depStatAcc_upd(&totalPartTableScanned, rsi->numPartScanned, rsh, rsi, nsi);
+		cdbexplain_depStatAcc_upd(&shareinputwait, rsi->shareinputwait, rsh, rsi, nsi);
 		if (rsi->sortMethod < NUM_SORT_METHOD && rsi->sortMethod != UNINITIALIZED_SORT && rsi->sortSpaceType != UNINITIALIZED_SORT_SPACE_TYPE)
 		{
 			Assert(rsi->sortSpaceType <= NUM_SORT_SPACE_TYPE);
@@ -1210,6 +1231,7 @@ cdbexplain_depositStatsToNode(PlanState *planstate, CdbExplain_RecvStatCtx *ctx)
 	ns->totalWorkfileCreated = totalWorkfileCreated.agg;
 	ns->peakMemBalance = peakMemBalance.agg;
 	ns->totalPartTableScanned = totalPartTableScanned.agg;
+	ns->shareinputwait = shareinputwait.agg;
 	for (int idx = 0; idx < NUM_SORT_METHOD; ++idx)
 	{
 		ns->sortSpaceUsed[MEMORY_SORT_SPACE_TYPE - 1][idx] = sortSpaceUsed[MEMORY_SORT_SPACE_TYPE - 1][idx].agg;
@@ -1887,6 +1909,51 @@ cdbexplain_showExecStats(struct PlanState *planstate, ExplainState *es)
 		}
 	}
 
+	/*
+	 * Print how long a cross-slice ShareInputScan blocked waiting for another
+	 * slice.
+	 *
+	 * This is aggregated rather than left to the per-worker extra message
+	 * text, because that text is selected by memory and row-count criteria
+	 * (see cdbexplain_depositStatsToNode): with a skewed wait it would report
+	 * an arbitrary worker rather than the one that actually blocked. Showing
+	 * the max, its segment, and the average makes the skew itself visible.
+	 *
+	 * Every cross-slice consumer pays some handshake cost, so only report a
+	 * wait that is material -- otherwise each such node would carry a
+	 * "0.000 ms" line of pure noise.
+	 */
+#define SHAREINPUT_WAIT_REPORT_THRESHOLD_SEC 0.001
+
+	if (T_ShareInputScanState == planstate->type &&
+		ns->shareinputwait.vmax >= SHAREINPUT_WAIT_REPORT_THRESHOLD_SEC)
+	{
+		double		wait_avg = cdbexplain_agg_avg(&ns->shareinputwait);
+
+		cdbexplain_formatSeg(segbuf, sizeof(segbuf), ns->shareinputwait.imax,
+							 ns->ninst);
+
+		if (es->format == EXPLAIN_FORMAT_TEXT)
+		{
+			appendStringInfoSpaces(es->str, es->indent * 2);
+			appendStringInfo(es->str,
+							 "Cross-slice wait: %.3f ms max%s, %.3f ms avg x %d workers.\n",
+							 ns->shareinputwait.vmax * 1000.0, segbuf,
+							 wait_avg * 1000.0, ns->shareinputwait.vcnt);
+		}
+		else
+		{
+			ExplainOpenGroup("cross-slice-wait", "cross-slice-wait", true, es);
+			ExplainPropertyFloat("Max Wait (ms)",
+								 ns->shareinputwait.vmax * 1000.0, 3, es);
+			ExplainPropertyInteger("Max Wait Segment",
+								   ns->shareinputwait.imax, es);
+			ExplainPropertyFloat("Avg Wait (ms)", wait_avg * 1000.0, 3, es);
+			ExplainPropertyInteger("Workers", ns->shareinputwait.vcnt, es);
+			ExplainCloseGroup("cross-slice-wait", "cross-slice-wait", true, es);
+		}
+	}
+
 	/*
 	 * Print number of partitioned tables scanned for dynamic scans.
 	 */

src/backend/executor/nodeShareInputScan.c

@@ -39,6 +39,7 @@
 #include "executor/executor.h"
 #include "executor/nodeShareInputScan.h"
 #include "miscadmin.h"
+#include "portability/instr_time.h"
 #include "utils/faultinjector.h"
 #include "utils/gp_alloc.h"
 #include "utils/tuplesort.h"
@@ -321,7 +322,20 @@ ExecSliceDependencyShareInputScan(ShareInputScanState *node)
 	EState *estate = node->ss.ps.state;
 	if(sisc->driver_slice >= 0 && sisc->driver_slice == currentSliceId)
 	{
-		shareinput_reader_waitready(node->share_lk_ctxt, sisc->share_id, estate->es_plannedstmt->planGen);
+		double		waited_sec;
+
+		shareinput_reader_waitready(node->share_lk_ctxt, sisc->share_id,
+									estate->es_plannedstmt->planGen,
+									&node->waitready_time);
+
+		/*
+		 * Roll the wait up to the query level so the stats collector can
+		 * report it (it only reads query-level, root-node instrumentation).
+		 * Keep the longest wait seen on this segment.
+		 */
+		waited_sec = INSTR_TIME_GET_DOUBLE(node->waitready_time);
+		if (waited_sec > estate->es_cross_slice_wait)
+			estate->es_cross_slice_wait = waited_sec;
 	}
 }
 
@@ -624,6 +638,41 @@ static void fi_close_created_fds(int *fds, char *file_prefix, int num)
  * that can cause deadlocks (OPT-2690).
  */
 
+/*
+ * Context for the errcontext callback installed while blocked on a
+ * cross-slice handshake. A query that never gets its handshake is killed by
+ * statement_timeout, so it produces no EXPLAIN ANALYZE output and no
+ * completion log line -- the elapsed wait would be lost entirely. Reporting
+ * it through errcontext puts it into the cancellation message itself, which
+ * is the only channel that survives.
+ */
+typedef struct shareinput_wait_errctx
+{
+	int			share_id;
+	int			slice_id;
+	bool		is_reader;
+	instr_time	starttime;
+} shareinput_wait_errctx;
+
+static void
+shareinput_wait_errcontext_callback(void *arg)
+{
+	shareinput_wait_errctx *ctx = (shareinput_wait_errctx *) arg;
+	instr_time	now;
+
+	INSTR_TIME_SET_CURRENT(now);
+	INSTR_TIME_SUBTRACT(now, ctx->starttime);
+
+	if (ctx->is_reader)
+		errcontext("ShareInputScan consumer (share_id=%d) in slice %d blocked %.3f ms waiting for the producer slice to publish",
+				   ctx->share_id, ctx->slice_id,
+				   INSTR_TIME_GET_MILLISEC(now));
+	else
+		errcontext("ShareInputScan producer (share_id=%d) in slice %d blocked %.3f ms waiting for consumer slices to finish reading",
+				   ctx->share_id, ctx->slice_id,
+				   INSTR_TIME_GET_MILLISEC(now));
+}
+
 /*
  * shareinput_reader_waitready
  *
@@ -633,12 +682,17 @@ static void fi_close_created_fds(int *fds, char *file_prefix, int num)
  *  This is a blocking operation.
  */
 void
-shareinput_reader_waitready(void *ctxt, int share_id, PlanGenerator planGen)
+shareinput_reader_waitready(void *ctxt, int share_id, PlanGenerator planGen,
+							instr_time *waited)
 {
 	struct pollfd fds[1];
 	int nfds = 0;
 	char a;
 	ShareInput_Lk_Context *pctxt = (ShareInput_Lk_Context *) ctxt;
+	instr_time	starttime;
+	instr_time	endtime;
+	shareinput_wait_errctx waitctx;
+	ErrorContextCallback errcallback;
 	RegisterXactCallbackOnce(XCallBack_ShareInput_FIFO, pctxt);
 
 #ifdef FAULT_INJECTOR
@@ -683,6 +737,18 @@ shareinput_reader_waitready(void *ctxt, int share_id, PlanGenerator planGen)
 	fds[0].fd = pctxt->readyfd;
 	fds[0].events = POLLIN;
 	nfds++;
+
+	INSTR_TIME_SET_CURRENT(starttime);
+
+	waitctx.share_id = share_id;
+	waitctx.slice_id = currentSliceId;
+	waitctx.is_reader = true;
+	waitctx.starttime = starttime;
+	errcallback.callback = shareinput_wait_errcontext_callback;
+	errcallback.arg = (void *) &waitctx;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
 	while(1)
 	{
 		CHECK_FOR_INTERRUPTS();
@@ -730,6 +796,14 @@ shareinput_reader_waitready(void *ctxt, int share_id, PlanGenerator planGen)
 					share_id, currentSliceId, save_errno);
 		}
 	}
+
+	error_context_stack = errcallback.previous;
+
+	INSTR_TIME_SET_CURRENT(endtime);
+	INSTR_TIME_ACCUM_DIFF(*waited, endtime, starttime);
+
+	elog(DEBUG1, "SISC READER (shareid=%d, slice=%d): waited %.3f ms for the producer",
+		 share_id, currentSliceId, INSTR_TIME_GET_MILLISEC(*waited));
 }
 
 /*
@@ -884,6 +958,10 @@ shareinput_writer_waitdone(void *ctxt, int share_id, int nsharer_xslice)
 	ShareInput_Lk_Context *pctxt = (ShareInput_Lk_Context *) ctxt;
 	struct pollfd fds[1];
 	int nfds = 0;
+	instr_time	starttime;
+	instr_time	waited;
+	shareinput_wait_errctx waitctx;
+	ErrorContextCallback errcallback;
 
 	if (pctxt->donefd < 0)
 		return;
@@ -897,6 +975,18 @@ shareinput_writer_waitdone(void *ctxt, int share_id, int nsharer_xslice)
 	fds[0].fd = pctxt->donefd;
 	fds[0].events = POLLIN;
 	nfds++;
+
+	INSTR_TIME_SET_CURRENT(starttime);
+
+	waitctx.share_id = share_id;
+	waitctx.slice_id = currentSliceId;
+	waitctx.is_reader = false;
+	waitctx.starttime = starttime;
+	errcallback.callback = shareinput_wait_errcontext_callback;
+	errcallback.arg = (void *) &waitctx;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
 	while(ack_needed > 0)
 	{
 		CHECK_FOR_INTERRUPTS();
@@ -931,8 +1021,14 @@ shareinput_writer_waitdone(void *ctxt, int share_id, int nsharer_xslice)
 		}
 	}
 
-	elog(DEBUG1, "SISC WRITER (shareid=%d, slice=%d): Writer received all %d reader done notifications",
-			share_id, currentSliceId, nsharer_xslice - pctxt->zcnt);
+	error_context_stack = errcallback.previous;
+
+	INSTR_TIME_SET_CURRENT(waited);
+	INSTR_TIME_SUBTRACT(waited, starttime);
+
+	elog(DEBUG1, "SISC WRITER (shareid=%d, slice=%d): Writer received all %d reader done notifications after %.3f ms",
+			share_id, currentSliceId, nsharer_xslice - pctxt->zcnt,
+			INSTR_TIME_GET_MILLISEC(waited));
 
 	shareinput_clean_lk_ctxt(ctxt);
 	UnregisterXactCallbackOnce(XCallBack_ShareInput_FIFO, (void *) ctxt);