You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by le...@apache.org on 2026/09/14 06:30:42 UTC
(cloudberry) branch REL_2_STABLE updated: Feature: add pg_query_state to gp_stats_collector (#1934)
This is an automated email from the ASF dual-hosted git repository.
leborchuk pushed a commit to branch REL_2_STABLE
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
The following commit(s) were added to refs/heads/REL_2_STABLE by this push:
new d27d84e8ac4 Feature: add pg_query_state to gp_stats_collector (#1934)
d27d84e8ac4 is described below
commit d27d84e8ac4f4a6297607da83bf135afe995a700
Author: Aleksey Rozhok <sh...@gmail.com>
AuthorDate: Mon Sep 14 09:30:36 2026 +0300
Feature: add pg_query_state to gp_stats_collector (#1934)
Adds a signal-based runtime query-state facility to the gp_stats_collector extension. It lets a session inspect the live execution state of another running backend on demand - walking its active plan tree across the QD and all QEs - without waiting for the query to finish, pushing batches to the UDS(unix domain socket).
New SQL API (extension v1.2, schema gpsc):
gpsc.pg_query_state(pid, trace_id) - fan out a poll to the query running on pid; each participating backend walks its plan tree and logs a per-node snapshot.
gpsc.pg_query_state_backends(pid) - list the (segid, pid) QE backends taking part in that query.
cbdb_mpp_query_state(gpsc.gp_segment_pid[], trace_id) - QE-side dispatch target.
The extension embeds the pg_query_state signal layer, which depends on three PostgreSQL core changes folded directly into the tree (configure enables the extension by default, so the tree must build without a manual patch step):
custom ProcSignal handlers (procsignal.c/.h, postgres.c);
end-of-node instrumentation flag readable mid-run (instrument.c/.h);
runtime EXPLAIN entry points (explain.c/.h).
---------
Co-authored-by: Dianjin Wang <wa...@gmail.com>
---
.github/workflows/build-cloudberry.yml | 4 +-
.github/workflows/build-deb-cloudberry.yml | 3 +-
.github/workflows/gpsc-crash-test.yaml | 278 +++++
LICENSE | 3 +
gpcontrib/gp_stats_collector/Makefile | 3 +-
gpcontrib/gp_stats_collector/README.md | 26 +
.../docs/pg_query_state_dataflow.puml | 113 ++
.../gp_stats_collector--1.1--1.2.sql | 49 +
.../gp_stats_collector/gp_stats_collector--1.2.sql | 159 +++
.../gp_stats_collector/gp_stats_collector.control | 2 +-
.../gp_stats_collector/protos/yagpcc_metrics.proto | 65 ++
.../gp_stats_collector/protos/yagpcc_plan.proto | 67 ++
.../protos/yagpcc_set_per_node.proto | 93 ++
.../gp_stats_collector/src/PlanNodeEmitter.cpp | 170 +++
.../src/{UDSConnector.h => PlanNodeEmitter.h} | 33 +-
gpcontrib/gp_stats_collector/src/UDSConnector.cpp | 153 ++-
gpcontrib/gp_stats_collector/src/UDSConnector.h | 18 +
.../gp_stats_collector/src/gp_stats_collector.c | 9 +
.../src/pg_query_state/README.md | 96 ++
.../src/pg_query_state/pg_query_state.c | 1191 ++++++++++++++++++++
.../src/pg_query_state/pg_query_state.h | 228 ++++
.../src/pg_query_state/qs_types.h | 112 ++
.../src/pg_query_state/signal_handler.c | 1000 ++++++++++++++++
gpcontrib/gp_stats_collector/test/Makefile | 22 +
gpcontrib/gp_stats_collector/test/crash/README.md | 128 +++
.../gp_stats_collector/test/crash/crash_scan.sh | 84 ++
.../test/crash/extract_failures.sh | 37 +
gpcontrib/gp_stats_collector/test/crash/poller.py | 149 +++
.../gp_stats_collector/test/crash/uds_drain.py | 75 ++
.../test/expected/gpsc_pg_query_state.out | 57 +
.../gp_stats_collector/test/isolation2/.gitignore | 4 +
.../gp_stats_collector/test/isolation2/Makefile | 43 +
.../test/isolation2/expected/gpsc_pqs_backends.out | 27 +
.../test/isolation2/expected/gpsc_pqs_disabled.out | 61 +
.../test/isolation2/expected/gpsc_pqs_perms.out | 92 ++
.../test/isolation2/expected/gpsc_pqs_running.out | 69 ++
.../isolation2/expected/gpsc_pqs_seg_count.out | 58 +
.../test/isolation2/expected/setup.out | 6 +
.../test/isolation2/isolation2_schedule | 20 +
.../test/isolation2/sql/gpsc_pqs_backends.sql | 21 +
.../test/isolation2/sql/gpsc_pqs_disabled.sql | 36 +
.../test/isolation2/sql/gpsc_pqs_perms.sql | 57 +
.../test/isolation2/sql/gpsc_pqs_running.sql | 45 +
.../test/isolation2/sql/gpsc_pqs_seg_count.sql | 36 +
.../test/isolation2/sql/setup.sql | 4 +
.../test/sql/gpsc_pg_query_state.sql | 46 +
licenses/LICENSE-pg_query_state.txt | 18 +
pom.xml | 3 +
src/backend/commands/explain.c | 156 ++-
src/backend/executor/instrument.c | 4 +
src/backend/storage/ipc/procsignal.c | 106 ++
src/backend/tcop/postgres.c | 8 +-
src/include/commands/explain.h | 2 +
src/include/executor/instrument.h | 3 +
src/include/storage/procsignal.h | 18 +-
55 files changed, 5323 insertions(+), 47 deletions(-)
diff --git a/.github/workflows/build-cloudberry.yml b/.github/workflows/build-cloudberry.yml
index 7abb658632d..f8b462c2c54 100644
--- a/.github/workflows/build-cloudberry.yml
+++ b/.github/workflows/build-cloudberry.yml
@@ -280,7 +280,9 @@ jobs:
"enable_core_check":false
},
{"test":"gpcontrib-gp-stats-collector",
- "make_configs":["gpcontrib/gp_stats_collector:installcheck"],
+ "make_configs":["gpcontrib/gp_stats_collector:installcheck",
+ "gpcontrib/gp_stats_collector/test:installcheck",
+ "gpcontrib/gp_stats_collector/test/isolation2:installcheck"],
"extension":"gp_stats_collector"
},
{"test":"gpcontrib-gp-relsizes-stats",
diff --git a/.github/workflows/build-deb-cloudberry.yml b/.github/workflows/build-deb-cloudberry.yml
index e4232cd4a04..230d884a30b 100644
--- a/.github/workflows/build-deb-cloudberry.yml
+++ b/.github/workflows/build-deb-cloudberry.yml
@@ -259,7 +259,8 @@ jobs:
"gpcontrib/gp_toolkit:installcheck"]
},
{"test":"gpcontrib-gp-stats-collector",
- "make_configs":["gpcontrib/gp_stats_collector:installcheck"],
+ "make_configs":["gpcontrib/gp_stats_collector:installcheck",
+ "gpcontrib/gp_stats_collector/test:installcheck"],
"extension":"gp_stats_collector"
},
{"test":"ic-cbdb-parallel",
diff --git a/.github/workflows/gpsc-crash-test.yaml b/.github/workflows/gpsc-crash-test.yaml
new file mode 100644
index 00000000000..50531f0a97b
--- /dev/null
+++ b/.github/workflows/gpsc-crash-test.yaml
@@ -0,0 +1,278 @@
+# --------------------------------------------------------------------
+#
+# 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.
+#
+# --------------------------------------------------------------------
+# gp_stats_collector crash test
+#
+# Proves that with the runtime query-state feature fully enabled and a poller
+# tracing every running query, Cloudberry does not crash and queries still
+# finish with the same results as without the feature.
+#
+# One build, one demo cluster, two regression passes on it:
+# run 1 baseline (feature OFF -- stock Cloudberry)
+# run 2 traced (feature ON + poller) + crash gate
+#
+# Hard verdict: the crash gate (no PANIC / signal / segment down / dead
+# coordinator). The failed-test delta (traced \ baseline) is reported for
+# information only -- the workload is not diff-deterministic -- and does not
+# fail the job.
+#
+# Workload: make installcheck-parallel (upstream parallel_schedule) -- fast and
+# fault-free, so any PANIC in the logs is a genuine crash.
+# --------------------------------------------------------------------
+name: GPSC Crash Test
+
+on:
+ push:
+ branches: [REL_2_STABLE]
+ pull_request:
+ branches: [REL_2_STABLE]
+ types: [opened, synchronize, reopened, edited]
+ workflow_dispatch:
+
+permissions:
+ contents: read
+
+concurrency:
+ group: ${{ github.workflow }}-${{ github.ref }}
+ cancel-in-progress: true
+
+jobs:
+ crash-test:
+ name: installcheck-parallel under tracing
+ runs-on: ubuntu-latest
+ container:
+ image: apache/incubator-cloudberry:cbdb-build-ubuntu22.04-latest
+ options: >-
+ --user root
+ -h cdw
+ env:
+ SRC_DIR: ${{ github.workspace }}/cloudberry
+ CRASH_DIR: ${{ github.workspace }}/cloudberry/gpcontrib/gp_stats_collector/test/crash
+ UDS_PATH: /tmp/gpsc_agent.sock
+ STOP_FILE: /tmp/gpsc_poller.stop
+
+ steps:
+ - name: Checkout Cloudberry source
+ uses: actions/checkout@v4
+ with:
+ path: cloudberry
+ submodules: recursive
+
+ - name: Cloudberry Environment Initialization
+ shell: bash
+ run: |
+ set -eo pipefail
+ if ! su - gpadmin -c "/tmp/init_system.sh"; then
+ echo "::error::Container initialization failed"
+ exit 1
+ fi
+ mkdir -p "${SRC_DIR}/build-logs"
+ chown -R gpadmin:gpadmin .
+ chmod -R 755 .
+
+ - name: Configure
+ shell: bash
+ run: |
+ set -eo pipefail
+ chmod +x "${SRC_DIR}"/devops/build/automation/cloudberry/scripts/configure-cloudberry.sh
+ if ! time su - gpadmin -c "cd ${SRC_DIR} && SRC_DIR=${SRC_DIR} ${SRC_DIR}/devops/build/automation/cloudberry/scripts/configure-cloudberry.sh"; then
+ echo "::error::Configure failed"
+ exit 1
+ fi
+
+ - name: Build
+ shell: bash
+ run: |
+ set -eo pipefail
+ chmod +x "${SRC_DIR}"/devops/build/automation/cloudberry/scripts/build-cloudberry.sh
+ if ! time su - gpadmin -c "cd ${SRC_DIR} && SRC_DIR=${SRC_DIR} ${SRC_DIR}/devops/build/automation/cloudberry/scripts/build-cloudberry.sh"; then
+ echo "::error::Build failed"
+ exit 1
+ fi
+
+ - name: Create stock demo cluster
+ shell: bash
+ run: |
+ set -eo pipefail
+ chmod +x "${SRC_DIR}"/devops/build/automation/cloudberry/scripts/create-cloudberry-demo-cluster.sh
+ if ! time su - gpadmin -c "cd ${SRC_DIR} && SRC_DIR=${SRC_DIR} ${SRC_DIR}/devops/build/automation/cloudberry/scripts/create-cloudberry-demo-cluster.sh"; then
+ echo "::error::Demo cluster creation failed"
+ exit 1
+ fi
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ gpconfig -c optimizer -v on && \
+ gpstop -ar"
+
+ - name: 'Run 1: baseline installcheck-parallel (feature OFF)'
+ shell: bash
+ run: |
+ set -eo pipefail
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ make -C src/test/regress installcheck-parallel > ${SRC_DIR}/build-logs/run1-baseline.log 2>&1" || true
+ cp -f "${SRC_DIR}/src/test/regress/regression.diffs" \
+ "${SRC_DIR}/build-logs/run1-baseline.diffs" 2>/dev/null || true
+ bash "${CRASH_DIR}/extract_failures.sh" "${SRC_DIR}/build-logs/run1-baseline.log" \
+ > "${SRC_DIR}/build-logs/baseline-failures.txt"
+ echo "baseline failures: $(wc -l < ${SRC_DIR}/build-logs/baseline-failures.txt)"
+ cat "${SRC_DIR}/build-logs/baseline-failures.txt"
+
+ - name: Reset state leaked by the baseline pass
+ shell: bash
+ run: |
+ set -eo pipefail
+ # installcheck recreates the 'regression' database each pass, but
+ # CREATE ROLE makes cluster-global roles that outlive it -- run 2's
+ # test_setup would then fail "role already exists". Drop the baseline
+ # database (clears role grants/ownership on it) and every regression-
+ # created role, so run 2 starts from the same clean slate as run 1.
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ psql -X -d postgres -c 'DROP DATABASE IF EXISTS regression;' && \
+ psql -X -q -A -t -d postgres \
+ -c \"SELECT format('DROP ROLE IF EXISTS %I;', rolname) FROM pg_roles WHERE rolname ~ '^(regress|mdb)'\" \
+ | psql -X -d postgres -f -"
+
+ - name: Enable full gp_stats_collector config
+ shell: bash
+ run: |
+ set -eo pipefail
+ # Phase 1: load the module, then restart so its custom GUCs are known.
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ gpconfig -c shared_preload_libraries -v 'gp_stats_collector' && \
+ gpstop -ar && \
+ sleep 10"
+ # Phase 2: enable every logging/polling knob, then restart again.
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ gpconfig -c pg_query_state.enable -v on && \
+ gpconfig -c pg_query_state.enable_timing -v on && \
+ gpconfig -c pg_query_state.enable_buffers -v on && \
+ gpconfig -c gpsc.enable -v on && \
+ gpconfig -c gpsc.enable_analyze -v on && \
+ gpconfig -c gpsc.enable_cdbstats -v on && \
+ gpconfig -c gpsc.report_nested_queries -v on && \
+ gpconfig -c gpsc.logging_mode -v UDS && \
+ gpconfig -c gpsc.uds_path -v ${UDS_PATH} && \
+ gpconfig -c compute_query_id -v regress && \
+ gpstop -ar && \
+ sleep 10"
+
+ - name: Install and smoke-test extension
+ shell: bash
+ run: |
+ set -eo pipefail
+ # The poller connects to 'postgres' and calls gpsc.pg_query_state; that
+ # SQL entry point only exists where the extension is created. Without
+ # this the traced run would be vacuous (every poll would just error on
+ # a missing function), so assert the function is resolvable and fail
+ # loudly if it is not.
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ psql -X -d postgres -c 'CREATE EXTENSION IF NOT EXISTS gp_stats_collector;' && \
+ psql -X -q -A -t -d postgres \
+ -c \"SELECT 'gpsc.pg_query_state(int,bytea)'::regprocedure;\"" \
+ || { echo "::error::gpsc.pg_query_state not resolvable -- extension not installed; traced run would be vacuous"; exit 1; }
+
+ - name: Start UDS drain
+ shell: bash
+ run: |
+ set -eo pipefail
+ chown -R gpadmin:gpadmin "${CRASH_DIR}"
+ su - gpadmin -c "nohup python3 ${CRASH_DIR}/uds_drain.py --path ${UDS_PATH} \
+ > ${SRC_DIR}/build-logs/uds-drain.log 2>&1 &"
+ sleep 2
+ test -S "${UDS_PATH}" || { echo "::error::UDS drain socket not created"; exit 1; }
+
+ - name: 'Run 2: traced installcheck-parallel (poller running)'
+ shell: bash
+ run: |
+ set -eo pipefail
+ rm -f "${STOP_FILE}"
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ nohup python3 ${CRASH_DIR}/poller.py --stop-file ${STOP_FILE} \
+ > ${SRC_DIR}/build-logs/poller.log 2>&1 &"
+ su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ make -C src/test/regress installcheck-parallel > ${SRC_DIR}/build-logs/run2-traced.log 2>&1" || true
+ touch "${STOP_FILE}"
+ sleep 3
+ cp -f "${SRC_DIR}/src/test/regress/regression.diffs" \
+ "${SRC_DIR}/build-logs/run2-traced.diffs" 2>/dev/null || true
+ bash "${CRASH_DIR}/extract_failures.sh" "${SRC_DIR}/build-logs/run2-traced.log" \
+ > "${SRC_DIR}/build-logs/traced-failures.txt"
+ echo "traced failures: $(wc -l < ${SRC_DIR}/build-logs/traced-failures.txt)"
+ cat "${SRC_DIR}/build-logs/traced-failures.txt"
+
+ - name: 'Crash gate (hard verdict)'
+ shell: bash
+ run: |
+ set -eo pipefail
+ if ! su - gpadmin -c "cd ${SRC_DIR} && \
+ source /usr/local/cloudberry-db/cloudberry-env.sh && \
+ source gpAux/gpdemo/gpdemo-env.sh && \
+ bash ${CRASH_DIR}/crash_scan.sh ${SRC_DIR}/gpAux/gpdemo/datadirs \
+ > ${SRC_DIR}/build-logs/crash-scan.log 2>&1"; then
+ echo "::error::Crash gate tripped -- Cloudberry did not survive tracing"
+ cat "${SRC_DIR}/build-logs/crash-scan.log"
+ exit 1
+ fi
+ cat "${SRC_DIR}/build-logs/crash-scan.log"
+
+ - name: 'Failed-test delta (informational)'
+ shell: bash
+ run: |
+ set -eo pipefail
+ # Tests whose tracing diff is client-message noise, not a correctness
+ # signal. strings: QD parse-time WARNINGs ("nonstandard use of \\",
+ # scan.l escape_string_warning) re-emit non-deterministically when the
+ # poller ProcSignal lands mid-statement; the query has no runtime stats.
+ printf '%s\n' strings | sort -u > "${SRC_DIR}/build-logs/known-flaky.txt"
+ # baseline/traced-failures.txt are already sort -u (extract_failures.sh).
+ comm -13 \
+ "${SRC_DIR}/build-logs/baseline-failures.txt" \
+ "${SRC_DIR}/build-logs/traced-failures.txt" \
+ | comm -23 - "${SRC_DIR}/build-logs/known-flaky.txt" \
+ > "${SRC_DIR}/build-logs/delta.txt"
+ count=$(wc -l < "${SRC_DIR}/build-logs/delta.txt")
+ echo "tests failing under tracing but not in the stock baseline: ${count}"
+ cat "${SRC_DIR}/build-logs/delta.txt"
+ if [ "${count}" -gt 0 ]; then
+ echo "::warning::${count} test(s) failed only under tracing (informational; the workload is not diff-deterministic -- inspect run2-traced.diffs)."
+ fi
+
+ - name: Upload crash-test artifacts
+ if: always()
+ uses: actions/upload-artifact@v4
+ with:
+ name: gpsc-crash-test-results
+ path: |
+ cloudberry/build-logs/
+ retention-days: 14
diff --git a/LICENSE b/LICENSE
index 5e3c7008094..be3387e74ac 100644
--- a/LICENSE
+++ b/LICENSE
@@ -344,6 +344,9 @@ Apache Cloudberry includes codes from
gpcontrib/yezzey/*
see licenses/LICENSE-yezzey.txt
+ gpcontrib/gp_stats_collector/src/pg_query_state/*
+ see licenses/LICENSE-pg_query_state.txt
+
----------------------------
Apache License - Version 2.0
diff --git a/gpcontrib/gp_stats_collector/Makefile b/gpcontrib/gp_stats_collector/Makefile
index b3228d2c45e..7c8e2b269af 100644
--- a/gpcontrib/gp_stats_collector/Makefile
+++ b/gpcontrib/gp_stats_collector/Makefile
@@ -3,7 +3,7 @@ EXTENSION = gp_stats_collector
DATA = $(wildcard *--*.sql)
REGRESS = gpsc_cursors gpsc_dist gpsc_select gpsc_utf8_trim gpsc_utility gpsc_guc_cache gpsc_uds gpsc_locale
-PROTO_BASES = gpsc_plan gpsc_metrics gpsc_set_service
+PROTO_BASES = gpsc_plan gpsc_metrics gpsc_set_service yagpcc_metrics yagpcc_plan yagpcc_set_per_node
PROTO_OBJS = $(patsubst %,src/protos/%.pb.o,$(PROTO_BASES))
C_OBJS = $(patsubst %.c,%.o,$(wildcard src/*.c src/*/*.c))
@@ -11,6 +11,7 @@ CPP_OBJS = $(patsubst %.cpp,%.o,$(wildcard src/*.cpp src/log/*.cpp src/memory/*.
OBJS = $(C_OBJS) $(CPP_OBJS) $(PROTO_OBJS)
PG_CXXFLAGS += -Werror -Wall -Wno-unused-but-set-variable -std=c++17 -Isrc/protos -Isrc -Iinclude -DGPBUILD
+PG_CPPFLAGS += -I$(libpq_srcdir) -Isrc/protos -Isrc -Iinclude
SHLIB_LINK += -lprotobuf -lstdc++
EXTRA_CLEAN = src/protos
diff --git a/gpcontrib/gp_stats_collector/README.md b/gpcontrib/gp_stats_collector/README.md
index 8c2d5c6868e..5b1ac0d1b25 100644
--- a/gpcontrib/gp_stats_collector/README.md
+++ b/gpcontrib/gp_stats_collector/README.md
@@ -45,3 +45,29 @@ An extension for collecting query execution metrics and reporting them to an ext
- **User Filtering:** To exclude activity from certain roles, add them to the comma-separated list in `gpsc.ignored_users_list`.
- **Trimming plans:** Query texts and execution plans are trimmed based on `gpsc.max_text_size` and `gpsc.max_plan_size` (default: 1024KB). For now, it is not recommended to set these GUCs higher than 1024KB.
- **Analyze collection:** Analyze is sent if execution time exceeds `gpsc.min_analyze_time`, which is 10 seconds by default. Analyze is collected if `gpsc.enable_analyze` is true.
+
+### Runtime Query State (`pg_query_state`)
+
+On-demand inspection of the live execution state of another running backend. The target's active plan tree is walked across the coordinator (QD) and every segment (QE), collecting per-node instrumentation, without waiting for the query to finish. Each backend pushes its own snapshot to the UDS sink configured by `gpsc.uds_path`, keyed by the caller-supplied `trace_id`.
+
+Delivery is best-effort, exactly like the rest of the extension: a snapshot that does not fit into the socket is dropped rather than retried, so a slow or absent reader never adds latency to the query being observed.
+
+The functions live in the `gpsc` schema (extension version 1.2).
+
+#### 1. `pg_query_state(pid, trace_id)`
+- **What:** Triggers runtime per-node collection for the query running on backend `pid`. Fans a poll out to every participating QE and to the QD; each backend walks its plan tree and pushes one per-node batch. The coordinator additionally pushes the deparsed plan document, rate-limited so that repeated polls of a long query do not resend an unchanged plan. Fire-and-forget: returns `void`.
+- **Arguments:** `trace_id` is a `bytea` of exactly 16 bytes, minted by the caller and used as the collection key on the receiving side.
+- **Executes on:** the coordinator only.
+- **GUC:** `pg_query_state.enable`.
+
+#### 2. `pg_query_state_backends(pid)`
+- **What:** Lists the QE backends participating in the query running on backend `pid`, as `(segid, pid)` rows, so that a collector knows how many batches to expect. A coordinator-only query (`INSERT ... VALUES`, catalog reads) allocates no gang, and is reported as a single row for the coordinator itself with `segid < 0`. Returns an empty set when the target is not running a query or has the module disabled.
+- **GUC:** `pg_query_state.enable`.
+
+#### 3. `cbdb_mpp_query_state(gp_segment_pid[], trace_id)`
+- **What:** QE-side dispatch target used internally by `pg_query_state()`; not intended for direct use.
+
+### Runtime Query State Configuration
+- **Enable:** `pg_query_state.enable` (default `on`) turns the executor hooks and signal handling on or off. Additional GUCs `pg_query_state.enable_timing` and `pg_query_state.enable_buffers` control the level of instrumentation collected.
+- **Permissions:** The functions are granted to `PUBLIC`, but access is checked in the server: a caller may poll a backend only if it is a superuser or owns the target query. This lets monitoring agents run under a non-superuser role while still preventing one role from observing another's queries.
+- **Preload:** The module registers custom signal handlers at startup, so `gp_stats_collector` must be listed in `shared_preload_libraries`.
diff --git a/gpcontrib/gp_stats_collector/docs/pg_query_state_dataflow.puml b/gpcontrib/gp_stats_collector/docs/pg_query_state_dataflow.puml
new file mode 100644
index 00000000000..bb0b2bc2340
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/docs/pg_query_state_dataflow.puml
@@ -0,0 +1,113 @@
+' 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 runtime per-node stats -- data flow.
+' Render: plantuml docs/pg_query_state_dataflow.puml (produces a PNG/SVG)
+'
+' Key point of this diagram: the SQL functions run on their OWN (requestor)
+' backends; the query being observed lives in SEPARATE (target) backends on the
+' same host, reached only by ProcSignal.
+
+@startuml pg_query_state_dataflow
+title pg_query_state — on-demand runtime per-node stats (keyed by trace_id)
+
+skinparam backgroundColor #FFFFFF
+skinparam shadowing false
+skinparam roundcorner 8
+skinparam sequence {
+ ArrowThickness 1.4
+ LifeLineBorderColor #9AA5B1
+ LifeLineBackgroundColor #F5F7FA
+ ParticipantBorderColor #52606D
+ ParticipantBackgroundColor #E4E7EB
+ ParticipantFontStyle bold
+ BoxBorderColor #B8C1CC
+ NoteBackgroundColor #FFF9E6
+ NoteBorderColor #E0C879
+}
+
+' Arrow colour convention (see legend):
+' blue = SQL / MPP dispatch (libpq)
+' red = ProcSignal (custom signal), same host
+' gray = shared-memory queue (shm_mq) reply
+' green = protobuf over Unix domain socket (UDS)
+
+actor UI as "UI"
+participant M as "yagpcc\n(master)"
+
+box "Coordinator host" #F0F4F8
+ participant RQD as "Requestor QD\nruns pg_query_state()"
+ participant TQD as "Target QD\n(observed query)"
+end box
+
+box "Segment hosts" #F0F4F8
+ collections RQE as "Requestor QEs\nrun cbdb_mpp_query_state()"
+ collections TQE as "Target QEs\n(observed query)"
+end box
+
+participant Y as "Local yagpcc\n(per-host UDS sink)"
+
+autonumber
+
+UI -[#2F80ED]> M : POST /api/per-node-stats?pid
+hnote over M #EAF2FF : mint **trace_id**\n(16 raw bytes, UUID)
+
+== Expected backend set (completeness barrier) ==
+M -[#2F80ED]> RQD : gpsc.pg_query_state_backends(pid)
+RQD -[#EB5757]> TQD : BackendInfoPollReason (signal)
+TQD -[#828282]-> RQD : active QE set via shm_mq\n(SendCdbComponents runs in TQD)
+RQD --[#2F80ED]> M : expected backend set (segid, pid)
+
+== Fan out the poll ==
+M -[#2F80ED]> RQD : gpsc.pg_query_state(pid, **trace_id**)
+note over RQD
+ validate **trace_id** (exactly 16 B)
+ permission gate:
+ superuser() || GetUserId() == proc->roleId
+ resolve target QEs (same BackendInfoPollReason round-trip)
+end note
+
+note over RQD, TQD : RQD stamps qs_trace_slots[**TQD** backendId] = **trace_id**
+RQD -[#EB5757]> TQD : SendProcSignal(QueryStatePollReason)
+
+RQD -[#2F80ED]> RQE : CdbDispatchCommand:\ncbdb_mpp_query_state(seg_pid[], **trace_id**)\n(a separate requestor backend per segment host)
+note over RQE, TQE : each RQE stamps qs_trace_slots[**TQE** backendId] = **trace_id**
+RQE -[#EB5757]> TQE : SendProcSignal(QueryStatePollReason)
+
+== Collect (in the target backends' signal handler) ==
+note over TQD, TQE
+ SendQueryState(): walk the LIVE plan tree,
+ one GpscNodeSample per plan node
+end note
+TQD -[#27AE60]> Y : SetPerNodeBatchReq (**trace_id**)
+TQE -[#27AE60]> Y : SetPerNodeBatchReq (**trace_id**)
+TQD -[#27AE60]> Y : SetQueryPlanReq (QD only, rate-limited)
+
+== Pull and pivot ==
+M -[#2F80ED]> Y : pull batches for **trace_id**\n(every segment, concurrent)
+Y --[#27AE60]> M : per-node batches
+hnote over M #EAF2FF : fold QD copy,\npivot flat samples\ninto a per-slice tree
+M --[#2F80ED]> UI : slices[] tree (JSON)\n(+ plan-doc via /api/per-node-plan)
+
+legend right
+ |= arrow |= transport |
+ | <color:#2F80ED>——</color> | SQL / MPP dispatch (libpq) |
+ | <color:#EB5757>——</color> | ProcSignal (same host) |
+ | <color:#828282>——</color> | shm_mq reply |
+ | <color:#27AE60>——</color> | protobuf over UDS |
+endlegend
+
+@enduml
diff --git a/gpcontrib/gp_stats_collector/gp_stats_collector--1.1--1.2.sql b/gpcontrib/gp_stats_collector/gp_stats_collector--1.1--1.2.sql
new file mode 100644
index 00000000000..ce0659d5032
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/gp_stats_collector--1.1--1.2.sql
@@ -0,0 +1,49 @@
+/* gp_stats_collector--1.1--1.2.sql */
+
+-- complain if script is sourced in psql, rather than via ALTER EXTENSION
+\echo Use "ALTER EXTENSION gp_stats_collector UPDATE TO '1.2'" to load this file. \quit
+
+-- 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;
+
+-- cbdb_mpp_query_state(gp_segment_pid[], trace_id): dispatched verbatim to
+-- every segment by pg_query_state() via CdbDispatchCommand; runs locally on
+-- each QE, so no EXECUTE ON marker. Signals the matching local backends. The
+-- trace_id is stamped into every per-node batch so all backends' pushes land
+-- under the one key this collection owns.
+CREATE FUNCTION gpsc.cbdb_mpp_query_state(gpsc.gp_segment_pid[], trace_id bytea)
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'cbdb_mpp_query_state'
+LANGUAGE C VOLATILE;
+
+-- pg_query_state_backends(pid): list the QE backends participating in the
+-- query running on backend `pid`, as (segid, pid) rows. yagpcc uses the row
+-- count as the "expected batches" barrier: per-node collection is complete
+-- once a batch has arrived from every listed backend.
+CREATE FUNCTION gpsc.pg_query_state_backends(pid int)
+RETURNS TABLE(segid int, pid int)
+AS 'MODULE_PATHNAME', 'pg_query_state_backends'
+LANGUAGE C VOLATILE EXECUTE ON COORDINATOR;
+
+-- The runtime query-state API is callable by any role; the per-backend
+-- permission gate in C (superuser or the query's owner) enforces access, so
+-- these can be granted broadly. This lets monitoring agents (e.g. yagpcc) run
+-- under a non-superuser role. cbdb_mpp_query_state is dispatched to the QEs
+-- under the caller's role, so it needs EXECUTE too.
+GRANT USAGE ON SCHEMA gpsc TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.pg_query_state(int, bytea) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.pg_query_state_backends(int) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.cbdb_mpp_query_state(gpsc.gp_segment_pid[], bytea) TO PUBLIC;
\ No newline at end of file
diff --git a/gpcontrib/gp_stats_collector/gp_stats_collector--1.2.sql b/gpcontrib/gp_stats_collector/gp_stats_collector--1.2.sql
new file mode 100644
index 00000000000..8e3bbeeae88
--- /dev/null
+++ b/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;
+
+-- cbdb_mpp_query_state(gp_segment_pid[], trace_id): dispatched verbatim to
+-- every segment by pg_query_state() via CdbDispatchCommand; runs locally on
+-- each QE, so no EXECUTE ON marker. Signals the matching local backends. The
+-- trace_id is stamped into every per-node batch so all backends' pushes land
+-- under the one key this collection owns.
+CREATE FUNCTION gpsc.cbdb_mpp_query_state(gpsc.gp_segment_pid[], trace_id bytea)
+RETURNS SETOF void
+AS 'MODULE_PATHNAME', 'cbdb_mpp_query_state'
+LANGUAGE C VOLATILE;
+
+-- pg_query_state_backends(pid): list the QE backends participating in the
+-- query running on backend `pid`, as (segid, pid) rows. yagpcc uses the row
+-- count as the "expected batches" barrier: per-node collection is complete
+-- once a batch has arrived from every listed backend.
+CREATE FUNCTION gpsc.pg_query_state_backends(pid int)
+RETURNS TABLE(segid int, pid int)
+AS 'MODULE_PATHNAME', 'pg_query_state_backends'
+LANGUAGE C VOLATILE EXECUTE ON COORDINATOR;
+
+-- The runtime query-state API is callable by any role; the per-backend
+-- permission gate in C (superuser or the query's owner) enforces access, so
+-- these can be granted broadly. This lets monitoring agents (e.g. yagpcc) run
+-- under a non-superuser role. cbdb_mpp_query_state is dispatched to the QEs
+-- under the caller's role, so it needs EXECUTE too.
+GRANT USAGE ON SCHEMA gpsc TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.pg_query_state(int, bytea) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.pg_query_state_backends(int) TO PUBLIC;
+GRANT EXECUTE ON FUNCTION gpsc.cbdb_mpp_query_state(gpsc.gp_segment_pid[], bytea) TO PUBLIC;
\ No newline at end of file
diff --git a/gpcontrib/gp_stats_collector/gp_stats_collector.control b/gpcontrib/gp_stats_collector/gp_stats_collector.control
index 4aea2bd49b8..76cf6c26e2b 100644
--- a/gpcontrib/gp_stats_collector/gp_stats_collector.control
+++ b/gpcontrib/gp_stats_collector/gp_stats_collector.control
@@ -1,5 +1,5 @@
# gp_stats_collector extension
comment = 'Intercept query and plan execution hooks and report them to Cloudberry monitor agents'
-default_version = '1.1'
+default_version = '1.2'
module_pathname = '$libdir/gp_stats_collector'
superuser = true
diff --git a/gpcontrib/gp_stats_collector/protos/yagpcc_metrics.proto b/gpcontrib/gp_stats_collector/protos/yagpcc_metrics.proto
new file mode 100644
index 00000000000..d51da308e44
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/protos/yagpcc_metrics.proto
@@ -0,0 +1,65 @@
+/**
+ * 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.
+ */
+
+syntax = "proto3";
+
+package yagpcc;
+
+/*
+ * Instrumentation counters for a single plan node execution.
+ *
+ * Field semantics mirror the PostgreSQL Instrumentation struct:
+ * ntuples -- total tuples produced (completed loops)
+ * nloops -- number of completed execution loops
+ * tuplecount -- tuples emitted so far in the current (in-progress) loop
+ * firsttuple -- wall time to first tuple of this cycle (seconds)
+ * startup -- total startup time across all loops (seconds)
+ * total -- total elapsed time across all loops (seconds)
+ */
+message MetricInstrumentation {
+ uint64 ntuples = 1;
+ uint64 nloops = 2;
+ uint64 tuplecount = 3;
+ double firsttuple = 4;
+ double startup = 5;
+ double total = 6;
+ uint64 shared_blks_hit = 7;
+ uint64 shared_blks_read = 8;
+}
+
+/*
+ * Node-level metrics container. Currently wraps only instrumentation; may
+ * be extended with system/spill stats in future revisions.
+ */
+message NodeMetrics {
+ MetricInstrumentation instrumentation = 1;
+}
+
+/*
+ * Common query and segment identification keys reused across messages.
+ */
+message QueryKey {
+ int32 tmid = 1; /* gp_gettmid() transaction/time identifier */
+ int32 ssid = 2; /* gp_session_id */
+ int32 ccnt = 3; /* gp_command_count */
+}
+
+message SegmentKey {
+ int32 dbid = 1; /* GpIdentity.dbid */
+ int32 segindex = 2; /* GpIdentity.segindex (-1 = coordinator) */
+}
diff --git a/gpcontrib/gp_stats_collector/protos/yagpcc_plan.proto b/gpcontrib/gp_stats_collector/protos/yagpcc_plan.proto
new file mode 100644
index 00000000000..3d38b5a50f7
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/protos/yagpcc_plan.proto
@@ -0,0 +1,67 @@
+/**
+ * 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.
+ */
+
+syntax = "proto3";
+
+import "google/protobuf/timestamp.proto";
+
+import "protos/yagpcc_metrics.proto";
+
+package yagpcc;
+
+message SetQueryPlanReq {
+ google.protobuf.Timestamp datetime = 1;
+ QueryKey query_key = 2;
+ string plan_doc = 3; /* ExplainPrintPlan output */
+ int32 format = 4; /* ExplainFormat: 0=text 1=xml 2=json 3=yaml */
+}
+
+/*
+ * Execution status of a single plan node.
+ *
+ * Mirrors QsNodeStatus from qs_types.h:
+ * INITIALIZED -- node was set up but has not yet started execution
+ * EXECUTING -- node is currently inside a tuple-fetch call
+ * FINISHED -- node has completed at least one full execution loop
+ */
+enum PlanNodeStatus {
+ PLAN_NODE_STATUS_UNSPECIFIED = 0;
+ PLAN_NODE_STATUS_INITIALIZED = 1;
+ PLAN_NODE_STATUS_EXECUTING = 2;
+ PLAN_NODE_STATUS_FINISHED = 3;
+}
+
+/*
+ * Identifying information for a single plan node within a query plan tree.
+ *
+ * Fields:
+ * plan_node_id -- unique node id within the plan (Plan.plan_node_id)
+ * parent_plan_node_id -- plan_node_id of the logical parent node, or 0
+ * node_type -- PostgreSQL NodeTag value (nodeTag(plan))
+ * slice_id -- CDB slice index (currentSliceId)
+ * plan_rows -- optimizer row-count estimate (Plan.plan_rows)
+ * relation_oid -- OID of the scanned relation for scan nodes, or 0
+ */
+message PlanNode {
+ int32 plan_node_id = 1;
+ int32 parent_plan_node_id = 2;
+ int32 node_type = 3;
+ int32 slice_id = 4;
+ double plan_rows = 5;
+ int32 relation_oid = 6;
+}
diff --git a/gpcontrib/gp_stats_collector/protos/yagpcc_set_per_node.proto b/gpcontrib/gp_stats_collector/protos/yagpcc_set_per_node.proto
new file mode 100644
index 00000000000..86746b2e4c4
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/protos/yagpcc_set_per_node.proto
@@ -0,0 +1,93 @@
+/**
+ * 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.
+ */
+
+syntax = "proto3";
+
+import "google/protobuf/timestamp.proto";
+
+import "protos/yagpcc_metrics.proto";
+import "protos/yagpcc_plan.proto";
+
+package yagpcc;
+
+/*
+ * SetPerNodeBatchReq -- one whole plan-tree snapshot from a single backend.
+ *
+ * Sent once per walker pass (SendQueryState signal or pg_qs_executor_end),
+ * carrying every observed plan node in one message. A backend opens one UDS
+ * connection and sends this batch instead of connect+send+close per node. The
+ * shared query_key, segment_key and datetime are hoisted out of every node.
+ *
+ * Wire transport: UDSConnector::report_per_node_batch() with the 8-byte
+ * extended protocol header (payload_size | 0x80000000, request_type=1,
+ * reserved=0).
+ *
+ * trace_id keys the whole collection: 16 raw bytes minted once on the
+ * coordinator per pg_query_state() call and stamped into every backend's batch,
+ * so yagpcc can group the snapshots of one poll and tell them apart from an
+ * overlapping poll of the same pid.
+ *
+ * This message MUST stay byte-identical to the yagpcc-side definition in
+ * api/proto/agent_segment/yagpcc_set_service.proto.
+ */
+message SetPerNodeBatchReq {
+ google.protobuf.Timestamp datetime = 1;
+ SegmentKey segment_key = 2;
+ repeated BatchNode nodes = 3;
+ bytes trace_id = 4; /* 16-byte per-collection key (see above) */
+}
+
+/*
+ * BatchNode -- one plan node inside a SetPerNodeBatchReq.
+ *
+ * Deliberately flat (no NodeMetrics wrapper) so the two repos can keep the
+ * message trivially wire-identical without sharing a metrics wrapper type.
+ * Fields mirror GpscNodeSample minus the hoisted identity keys.
+ */
+message BatchNode {
+ int32 plan_node_id = 1;
+ int32 parent_plan_node_id = 2;
+ int32 node_type = 3; /* raw NodeTag value */
+ int32 slice_id = 4;
+ double plan_rows = 5; /* planner estimate */
+ int32 relation_oid = 6; /* scan relation OID, 0 otherwise */
+ double ntuples = 7;
+ double tuplecount = 8;
+ double nloops = 9;
+ double startup = 10;
+ double total = 11;
+ double firsttuple = 12;
+ uint64 shared_blks_hit = 13;
+ uint64 shared_blks_read = 14;
+ PlanNodeStatus node_status = 15;
+ bool eof = 16;
+ google.protobuf.Timestamp executed_at = 17;
+ bool workfile_created = 18;
+ int64 workmem_used = 19; /* bytes of work_mem actually used */
+ int64 workmem_wanted = 20; /* bytes needed to avoid spill; >0 == spilled */
+ double ntuples_delta = 21; /* tuples produced since the previous sample */
+ double tuples_per_sec = 22; /* ntuples_delta over the sample interval */
+ double time_since_init_sec = 23; /* seconds since the node's first sample */
+ bool stalled = 24; /* executing, no new tuples, not at eof */
+ int32 pid = 25;
+ reserved 26;
+ reserved "segindex";
+ string relation_name = 27;
+ int32 ccnt = 28;
+}
+
diff --git a/gpcontrib/gp_stats_collector/src/PlanNodeEmitter.cpp b/gpcontrib/gp_stats_collector/src/PlanNodeEmitter.cpp
new file mode 100644
index 00000000000..4e299e3453a
--- /dev/null
+++ b/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;
+
+ yagpcc::SetPerNodeBatchReq request;
+
+ /* Timestamp */
+ *request.mutable_datetime() = current_ts();
+ request.set_trace_id(trace_id, GPSC_TRACE_ID_LEN);
+
+ auto *sk = request.mutable_segment_key();
+ sk->set_dbid(nodes[0]->dbid);
+ sk->set_segindex(nodes[0]->segindex);
+
+ for (int i = 0; i < count; i++)
+ {
+ GpscNodeSample *node = nodes[i];
+ yagpcc::BatchNode *bn = request.add_nodes();
+
+ bn->set_pid(node->pid);
+ bn->set_plan_node_id(node->plan_node_id);
+ bn->set_parent_plan_node_id(node->parent_plan_node_id);
+ bn->set_node_type(node->node_tag);
+ bn->set_slice_id(node->slice_id);
+ bn->set_plan_rows(node->plan_rows);
+ bn->set_relation_oid(node->relation_oid);
+ bn->set_ntuples(node->ntuples);
+ bn->set_tuplecount(node->tuplecount);
+ bn->set_nloops(node->nloops);
+ bn->set_startup(node->startup);
+ bn->set_total(node->total);
+ bn->set_firsttuple(node->firsttuple);
+ bn->set_shared_blks_hit(node->shared_blks_hit);
+ bn->set_shared_blks_read(node->shared_blks_read);
+ bn->set_node_status(map_node_status(node->node_status));
+ bn->set_eof(node->eof);
+ bn->set_relation_name(node->relation_name);
+ bn->set_ccnt(node->ccnt);
+ /*
+ * executed_at is the snapshot instant, shared by every node in this
+ * pass. It is stamped per node (not at message level) because the
+ * receiver aggregates nodes across segments and loses the batch
+ * grouping; each node needs its own compute time to derive a per-node
+ * rate. Same value as datetime here since one walk = one instant.
+ */
+ *bn->mutable_executed_at() = request.datetime();
+ bn->set_workfile_created(node->workfile_created);
+ bn->set_workmem_used(node->workmem_used);
+ bn->set_workmem_wanted(node->workmem_wanted);
+ /*
+ * Derived rate fields, computed in signal_handler from the per-node
+ * rolling state (prev ntuples + prev executed_at). They MUST be
+ * serialized here too: the receiver keys per invocation trace_id and
+ * sees each node once, so it cannot re-derive a rate on its side.
+ */
+ bn->set_ntuples_delta(node->ntuples_delta);
+ bn->set_tuples_per_sec(node->tuples_per_sec);
+ bn->set_time_since_init_sec(node->time_since_init_sec);
+ bn->set_stalled(node->stalled);
+ }
+
+ UDSConnector::report_per_node_batch(request, pne_config);
+}
+
+extern "C" void
+gpsc_emit_query_plan(int32_t tmid, int32_t ssid, int32_t ccnt,
+ const char *plan_doc, int32_t format)
+{
+ if (plan_doc == nullptr || plan_doc[0] == '\0')
+ return;
+
+ yagpcc::SetQueryPlanReq request;
+
+ *request.mutable_datetime() = current_ts();
+
+ auto *qk = request.mutable_query_key();
+ qk->set_tmid(tmid);
+ qk->set_ssid(ssid);
+ qk->set_ccnt(ccnt);
+
+ request.set_plan_doc(plan_doc);
+ request.set_format(format);
+
+ UDSConnector::report_query_plan(request, pne_config);
+}
diff --git a/gpcontrib/gp_stats_collector/src/UDSConnector.h b/gpcontrib/gp_stats_collector/src/PlanNodeEmitter.h
similarity index 63%
copy from gpcontrib/gp_stats_collector/src/UDSConnector.h
copy to gpcontrib/gp_stats_collector/src/PlanNodeEmitter.h
index ac56dd54f44..64610353ab1 100644
--- a/gpcontrib/gp_stats_collector/src/UDSConnector.h
+++ b/gpcontrib/gp_stats_collector/src/PlanNodeEmitter.h
@@ -17,26 +17,29 @@
* specific language governing permissions and limitations
* under the License.
*
- * UDSConnector.h
+ * PlanNodeEmitter.h
*
* IDENTIFICATION
- * gpcontrib/gp_stats_collector/src/UDSConnector.h
- *
- *-------------------------------------------------------------------------
+ * gpcontrib/gp_stats_collector/src/PlanNodeEmitter.h
*/
-#ifndef UDSCONNECTOR_H
-#define UDSCONNECTOR_H
+#ifndef PLAN_NODE_EMITTER_H
+#define PLAN_NODE_EMITTER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
-#include "protos/gpsc_set_service.pb.h"
+#include "pg_query_state/qs_types.h"
-class Config;
+extern void gpsc_emit_node_batch(GpscNodeSample **nodes, int count,
+ const char *trace_id);
+extern void gpsc_emit_query_plan(int32_t tmid, int32_t ssid, int32_t ccnt,
+ const char *plan_doc, int32_t format);
+extern void gpsc_qs_sync_config();
-class UDSConnector
-{
-public:
- bool static report_query(const gpsc::SetQueryReq &req,
- const std::string &event, const Config &config);
-};
+#ifdef __cplusplus
+}
+#endif
-#endif /* UDSCONNECTOR_H */
+#endif /* PLAN_NODE_EMITTER_H */
diff --git a/gpcontrib/gp_stats_collector/src/UDSConnector.cpp b/gpcontrib/gp_stats_collector/src/UDSConnector.cpp
index 056fa9071a5..6346ba5275d 100644
--- a/gpcontrib/gp_stats_collector/src/UDSConnector.cpp
+++ b/gpcontrib/gp_stats_collector/src/UDSConnector.cpp
@@ -28,8 +28,8 @@
#include "UDSConnector.h"
#include "Config.h"
#include "GpscStat.h"
-#include "log/LogOps.h"
#include "memory/gpdbwrappers.h"
+#include "pg_query_state/qs_types.h"
#include <string>
#include <sys/fcntl.h>
@@ -142,3 +142,154 @@ UDSConnector::report_query(const gpsc::SetQueryReq &req,
GpscStat::report_send(total_size);
return true;
}
+
+// Extended protocol used by the runtime query-state messages. The high bit of
+// the size word tells the receiver that an 8-byte header follows instead of the
+// original 4-byte one; the request type word then selects the payload message.
+static const uint32_t kExtendedProtocolFlag = 0x80000000u;
+static const uint16_t kRequestTypePerNodeBatch = 1;
+static const uint16_t kRequestTypeQueryPlan = 2;
+
+static void inline log_tracing_failure(const yagpcc::SetPerNodeBatchReq &req)
+{
+ static const char hexchars[] = "0123456789abcdef";
+ const unsigned char *trace_id =
+ reinterpret_cast<const unsigned char *>(req.trace_id().data());
+ size_t len = req.trace_id().size();
+ char hex[GPSC_TRACE_ID_LEN * 2 + 1];
+
+ if (len > GPSC_TRACE_ID_LEN)
+ len = GPSC_TRACE_ID_LEN;
+
+ for (size_t i = 0; i < len; ++i)
+ {
+ hex[i * 2] = hexchars[trace_id[i] >> 4];
+ hex[i * 2 + 1] = hexchars[trace_id[i] & 0x0f];
+ }
+ hex[len * 2] = '\0';
+
+ ereport(LOG,
+ (errmsg("Per-node batch {%s} tracing of %d nodes failed with error %m",
+ hex, req.nodes_size())));
+}
+
+static void inline log_tracing_failure(const yagpcc::SetQueryPlanReq &req)
+{
+ ereport(LOG,
+ (errmsg("Query {%d-%d-%d} plan document of %zu bytes failed with error %m",
+ req.query_key().tmid(), req.query_key().ssid(),
+ req.query_key().ccnt(), req.plan_doc().size())));
+}
+
+// Sends req behind the 8-byte extended header. Delivery repeats report_query()
+// above: a fresh non-blocking socket per message, MSG_DONTWAIT, a nap between
+// packets, and a plain drop once the socket refuses to take more. We never wait
+// on the reader -- a stalled yagpcc must not slow down the query it observes.
+//
+// The template exists so that log_tracing_failure() resolves by overload while
+// still being called before ~SockGuard() closes the socket and clobbers errno.
+template <typename Req>
+static bool
+report_extended(const Req &req, uint16_t request_type, const Config &config)
+{
+ sockaddr_un address{};
+ address.sun_family = AF_UNIX;
+ const auto &uds_path = config.uds_path();
+
+ if (uds_path.size() >= sizeof(address.sun_path))
+ {
+ ereport(WARNING, (errmsg("UDS path is too long for socket buffer")));
+ GpscStat::report_error();
+ return false;
+ }
+ strcpy(address.sun_path, uds_path.c_str());
+
+ const auto sockfd = socket(AF_UNIX, SOCK_STREAM, 0);
+ if (sockfd == -1)
+ {
+ log_tracing_failure(req);
+ GpscStat::report_error();
+ return false;
+ }
+
+ // Close socket automatically on error path.
+ struct SockGuard
+ {
+ int fd;
+ ~SockGuard()
+ {
+ close(fd);
+ }
+ } sock_guard{sockfd};
+
+ if (fcntl(sockfd, F_SETFL, O_NONBLOCK) == -1)
+ {
+ // That's a very important error that should never happen, so make it
+ // visible to an end-user and admins.
+ ereport(WARNING,
+ (errmsg("Unable to create non-blocking socket connection %m")));
+ GpscStat::report_error();
+ return false;
+ }
+
+ if (connect(sockfd, reinterpret_cast<sockaddr *>(&address),
+ sizeof(address)) == -1)
+ {
+ log_tracing_failure(req);
+ GpscStat::report_bad_connection();
+ return false;
+ }
+
+ const auto data_size = req.ByteSizeLong();
+ const auto header_size = sizeof(uint32_t) + 2 * sizeof(uint16_t);
+ const auto total_size = data_size + header_size;
+ auto *buf = static_cast<uint8_t *>(gpdb::palloc(total_size));
+ struct BufGuard
+ {
+ void *p;
+ ~BufGuard()
+ {
+ gpdb::pfree(p);
+ }
+ } buf_guard{buf};
+
+ *reinterpret_cast<uint32_t *>(buf) =
+ static_cast<uint32_t>(data_size) | kExtendedProtocolFlag;
+ *reinterpret_cast<uint16_t *>(buf + sizeof(uint32_t)) = request_type;
+ *reinterpret_cast<uint16_t *>(buf + sizeof(uint32_t) + sizeof(uint16_t)) = 0;
+ req.SerializeWithCachedSizesToArray(buf + header_size);
+
+ int64_t sent = 0, sent_total = 0;
+ do
+ {
+ sent = send(sockfd, buf + sent_total, total_size - sent_total,
+ MSG_DONTWAIT);
+ if (sent > 0)
+ sent_total += sent;
+ } while (sent > 0 && size_t(sent_total) != total_size &&
+ (pg_usleep(1000), true));
+
+ if (sent < 0)
+ {
+ log_tracing_failure(req);
+ GpscStat::report_bad_send(total_size);
+ return false;
+ }
+
+ GpscStat::report_send(total_size);
+ return true;
+}
+
+bool
+UDSConnector::report_per_node_batch(const yagpcc::SetPerNodeBatchReq &req,
+ const Config &config)
+{
+ return report_extended(req, kRequestTypePerNodeBatch, config);
+}
+
+bool
+UDSConnector::report_query_plan(const yagpcc::SetQueryPlanReq &req,
+ const Config &config)
+{
+ return report_extended(req, kRequestTypeQueryPlan, config);
+}
diff --git a/gpcontrib/gp_stats_collector/src/UDSConnector.h b/gpcontrib/gp_stats_collector/src/UDSConnector.h
index ac56dd54f44..4258617b190 100644
--- a/gpcontrib/gp_stats_collector/src/UDSConnector.h
+++ b/gpcontrib/gp_stats_collector/src/UDSConnector.h
@@ -29,6 +29,8 @@
#define UDSCONNECTOR_H
#include "protos/gpsc_set_service.pb.h"
+#include "protos/yagpcc_plan.pb.h"
+#include "protos/yagpcc_set_per_node.pb.h"
class Config;
@@ -37,6 +39,22 @@ class UDSConnector
public:
bool static report_query(const gpsc::SetQueryReq &req,
const std::string &event, const Config &config);
+
+ // The two calls below use the extended 8-byte header:
+ // bytes 0-3: payload_size | 0x80000000 (uint32)
+ // bytes 4-5: request type (uint16)
+ // bytes 6-7: reserved, zero (uint16)
+ // bytes 8+: serialized message
+ // Delivery is the same best-effort push as report_query(): the message is
+ // dropped when the socket cannot take it.
+
+ // Sends a whole plan-tree snapshot of one backend, request type 1.
+ bool static report_per_node_batch(const yagpcc::SetPerNodeBatchReq &req,
+ const Config &config);
+
+ // Sends the coordinator-only deparsed plan document, request type 2.
+ bool static report_query_plan(const yagpcc::SetQueryPlanReq &req,
+ const Config &config);
};
#endif /* UDSCONNECTOR_H */
diff --git a/gpcontrib/gp_stats_collector/src/gp_stats_collector.c b/gpcontrib/gp_stats_collector/src/gp_stats_collector.c
index d295e37b396..686159e0c3d 100644
--- a/gpcontrib/gp_stats_collector/src/gp_stats_collector.c
+++ b/gpcontrib/gp_stats_collector/src/gp_stats_collector.c
@@ -31,6 +31,7 @@
#include "utils/builtins.h"
#include "hook_wrappers.h"
+#include "pg_query_state/pg_query_state.h"
PG_MODULE_MAGIC;
@@ -50,6 +51,14 @@ _PG_init(void)
{
if (Gp_role == GP_ROLE_DISPATCH || Gp_role == GP_ROLE_EXECUTE)
hooks_init();
+
+ /*
+ * pg_query_state registers its own shared memory, ProcSignal handlers and
+ * executor hooks. It goes last on purpose: hooks are chained head-first,
+ * so registering after hooks_init() puts it outside of the collector's
+ * executor wrappers, which is what it needs to see an untouched QueryDesc.
+ */
+ pg_qs_init();
}
void
diff --git a/gpcontrib/gp_stats_collector/src/pg_query_state/README.md b/gpcontrib/gp_stats_collector/src/pg_query_state/README.md
new file mode 100644
index 00000000000..8b0f3053c0a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/src/pg_query_state/README.md
@@ -0,0 +1,96 @@
+<!--
+ 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 signal layer
+
+On-demand inspection of a *running* query: on request, every backend executing
+the query (the coordinator and all its QEs) walks its live plan tree and reports
+per-node runtime stats, without waiting for the query to finish.
+
+The code here is derived from [pg_query_state](https://github.com/postgrespro/pg_query_state)
+(PostgreSQL License) but the transport and keying differ substantially, so this
+note describes the design as implemented in this tree, not the upstream one.
+
+## Files in this directory
+
+- `pg_query_state.c` — `_PG_init` wiring, GUCs, executor hooks, the SQL entry
+ points, the permission gate, and the per-backend `qs_trace_slots` shmem.
+- `signal_handler.c` — the two custom ProcSignal handlers, the plan-tree walker,
+ the per-node delta computation, and the plan-doc builder.
+- `qs_types.h` — `GpscNodeSample` (one per-node sample) and the status enums.
+
+The C++ side that serializes and ships the samples lives one level up in `../`
+(`PlanNodeEmitter`, `UDSConnector`, `ProtoUtils`); the wire messages are in
+`../../protos/`. The receiver is yagpcc; its side is in `../../../../../yagpcc`.
+
+## The key: trace_id
+
+Every collection is keyed by a **trace_id** — 16 raw bytes minted once, on the
+coordinator, per `pg_query_state()` call. It is threaded to every participating
+backend and stamped into every message. yagpcc groups the snapshots of one poll
+by trace_id, and uses it to tell an in-flight poll apart from an overlapping poll
+of the same pid. `(tmid, ssid, ccnt)` are display-only.
+
+## Data flow
+
+The sequence diagram lives in
+[`../../docs/pg_query_state_dataflow.puml`](../../docs/pg_query_state_dataflow.puml)
+(render with `plantuml docs/pg_query_state_dataflow.puml`). In short:
+
+1. **Trigger.** yagpcc mints the trace_id and calls `gpsc.pg_query_state(pid,
+ trace_id)` on the coordinator (`EXECUTE ON COORDINATOR`). This runs on a fresh
+ *requestor* backend, not the backend running the observed query. yagpcc
+ separately calls `gpsc.pg_query_state_backends(pid)` to learn the exact set of
+ backends to expect — the completeness barrier for the pull below.
+2. **Fan-out.** The requestor validates the trace_id and checks the permission
+ gate, then resolves the observed query's QEs by signalling its coordinator
+ backend (`BackendInfoPollReason`; `SendCdbComponents` replies over shm_mq). For
+ each *target* backend it stamps `qs_trace_slots[backendId] = trace_id` and then
+ signals it with `QueryStatePollReason`: the target QD directly, the target QEs
+ via a dispatched `cbdb_mpp_query_state(gp_segment_pid[], trace_id)` that runs on
+ a requestor backend on each segment host. Requestor and target always sit on
+ the same host and only ever talk by signal.
+3. **Collection.** `SendQueryState()` runs in the signalled *target* backend. It
+ walks the live plan tree, builds one `GpscNodeSample` per node, and pushes the
+ whole snapshot as a single `SetPerNodeBatchReq` over UDS to the *local* yagpcc
+ (one connection per backend, not per node). The coordinator additionally builds
+ the deparsed plan document and sends it as a `SetQueryPlanReq`, rate-limited so
+ repeated polls of a long query do not resend an unchanged plan.
+4. **Storage & pull.** yagpcc stores each batch keyed by trace_id. The master
+ pulls every segment's batch for that trace, folds in the QD's own nodes, pivots
+ the flat samples into a per-slice tree, and returns it to the UI.
+
+## What a node sample carries
+
+`GpscNodeSample` (see `qs_types.h`) per plan node: identity (`plan_node_id`,
+`parent`, `slice_id`, `node_type`, scan `relation_oid`), counters (`ntuples`,
+`tuplecount`, `nloops`), timing (`startup`, `total`, `firsttuple`), buffers
+(`shared_blks_hit/read`), spill (`workmem_used/wanted`, `workfile_created`),
+status (`INITIALIZED`/`EXECUTING`/`FINISHED`, `eof`), and C-side derived rates
+(`ntuples_delta`, `tuples_per_sec`, `time_since_init_sec`, `stalled`). The rates
+come from a per-node rolling state keyed by `plan_node_id`, reset on executor
+start and end. The walk root reports `parent = -1`
+(`GPSC_NO_PARENT_PLAN_NODE_ID`); `0` would be ambiguous, since `plan_node_id`
+counters start there.
+
+## Permissions
+
+The SQL functions are granted to `PUBLIC` so a non-superuser monitoring agent
+can run them; access is gated in C: a caller may poll a backend only if it is a
+superuser or owns the target query (`GetUserId() == proc->roleId`).
diff --git a/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.c b/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.c
new file mode 100644
index 00000000000..1648261526a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.c
@@ -0,0 +1,1191 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ */
+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);
+}
+
+/*
+ * 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);
+ if (sig_result == -1)
+ ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("cbdb_mpp_query_state: failed to send signal to pid %d",
+ proc->pid)));
+ }
+ PG_RETURN_VOID();
+}
+
+/*
+ * get_query_backend_info -- convert a gp_segment_pid[] SQL array to a list
+ * of PGPROC pointers for backends running on this segment.
+ *
+ * Skips entries for other segments and entries whose backend has exited.
+ */
+static List *
+get_query_backend_info(ArrayType *array)
+{
+ int16 typlen;
+ bool typbyval;
+ char typalign;
+ Oid element_type = ARR_ELEMTYPE(array);
+ Datum *data;
+ bool *nulls;
+ int nitems;
+ int len;
+ List *alive_procs = NIL;
+
+ get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);
+ deconstruct_array(array, element_type, typlen, typbyval, typalign,
+ &data, &nulls, &nitems);
+
+ len = ArrayGetNItems(ARR_NDIM(array), ARR_DIMS(array));
+
+ for (int i = 0; i < len; i++)
+ {
+ if (nulls[i])
+ continue;
+
+ HeapTupleHeader td = DatumGetHeapTupleHeader(data[i]);
+ TupleDesc tupdesc;
+ HeapTupleData tmptup;
+ int32 pid;
+ int32 segid;
+ bool segid_isnull = false;
+ bool pid_isnull = false;
+ PGPROC *proc;
+
+ tupdesc = lookup_rowtype_tupdesc_copy(
+ HeapTupleHeaderGetTypeId(td), HeapTupleHeaderGetTypMod(td));
+ tmptup.t_len = HeapTupleHeaderGetDatumLength(td);
+ tmptup.t_data = td;
+
+ segid = DatumGetInt32(heap_getattr(&tmptup, 1, tupdesc, &segid_isnull));
+ pid = DatumGetInt32(heap_getattr(&tmptup, 2, tupdesc, &pid_isnull));
+ FreeTupleDesc(tupdesc);
+
+ if (segid_isnull || pid_isnull || segid != GpIdentity.segindex)
+ continue;
+
+ proc = BackendPidGetProc(pid);
+ if (!proc)
+ continue;
+
+ alive_procs = lappend(alive_procs, proc);
+ }
+ return alive_procs;
+}
diff --git a/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.h b/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.h
new file mode 100644
index 00000000000..4f532631958
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/src/pg_query_state/pg_query_state.h
@@ -0,0 +1,228 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.h
+ * Public API for the pg_query_state signal-dispatch layer.
+ *
+ * This header is included by the C extension entry point (gp_stats_collector.c),
+ * which only has to call pg_qs_init(); everything else the module needs it
+ * registers itself. Keep it C-compatible: no C++ types, wrapped in extern "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.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef __PG_QUERY_STATE_H__
+#define __PG_QUERY_STATE_H__
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "postgres.h"
+
+#include "commands/explain.h"
+#include "nodes/pg_list.h"
+#include "storage/procarray.h"
+#include "storage/shm_mq.h"
+#include "cdb/cdbdispatchresult.h"
+#include "qs_types.h"
+
+/* Shared memory queue capacity for passing query-state messages. */
+#define QUEUE_SIZE (64 * 1024)
+
+/* Maximum single chunk size when splitting a message across shm_mq sends. */
+#define MSG_MAX_SIZE (4 * 1024)
+
+/* Delay between shm_mq send retries, in microseconds (100 ms). */
+#define WRITING_DELAY (100 * 1000)
+
+/* Maximum number of send retries before giving up. */
+#define NUM_OF_ATTEMPTS 6
+
+/* Bitmask flags for caller-side warnings embedded in shm_mq_msg.warnings. */
+#define TIMING_OFF_WARNING 1
+#define BUFFERS_OFF_WARNING 2
+
+/* Unique key that identifies our shm_toc segment. */
+#define PG_QS_MODULE_KEY 0xCA94B108
+
+/* Table-of-contents slot indices within the shm_toc segment. */
+#define PG_QS_RCV_KEY 0
+#define PG_QS_SND_KEY 1
+
+/*
+ * Timeouts for shm_mq operations.
+ * The receive timeout must exceed the send timeout so that waiting workers
+ * always give up before the polling process stops listening.
+ */
+#define MAX_RCV_TIMEOUT 2000 /* ms */
+#define MAX_SND_TIMEOUT 1000 /* ms */
+
+/*
+ * Sleep between partial-receive retries (SHM_MQ_WOULD_BLOCK case).
+ * Must be less than MAX_RCV_TIMEOUT.
+ */
+#define PART_RCV_DELAY 100 /* ms */
+
+/*
+ * Minimum interval between coordinator plan-doc pushes for the same query.
+ * SendQueryState() re-sends the ExplainPrintPlan document only after this
+ * interval elapses, so repeated polls of a long-running query do not resend
+ * the (unchanging) plan on every signal.
+ */
+#define PLAN_DOC_RESEND_INTERVAL_MS (2 * 60 * 1000)
+
+/*
+ * Status codes returned by the signal handler to describe the state of the
+ * queried backend.
+ */
+typedef enum
+{
+ QUERY_NOT_RUNNING, /* backend is idle or has no active QueryDesc */
+ STAT_DISABLED, /* pg_query_state.enable = false */
+ QS_RETURNED, /* handler successfully collected and sent stats */
+ WRONG_ROLE /* target is a QE, not the QD: only GP_ROLE_DISPATCH
+ * knows the participant list, so no other backend can
+ * answer BackendInfoPollReason */
+} PG_QS_RequestResult;
+
+/*
+ * Wire format for a query-state reply message transmitted through shm_mq.
+ * The variable-length `stack` field carries sequentially laid out text frames,
+ * one per stack depth.
+ */
+typedef struct
+{
+ int reqid;
+ int length; /* total message size including flexible array */
+ PGPROC *proc;
+ PG_QS_RequestResult result_code;
+ int warnings; /* bitmask of TIMING_OFF_WARNING / BUFFERS_OFF_WARNING */
+ int stack_depth;
+ char stack[FLEXIBLE_ARRAY_MEMBER];
+} shm_mq_msg;
+
+#define BASE_SIZEOF_SHM_MQ_MSG (offsetof(shm_mq_msg, stack_depth))
+
+/*
+ * Compact identifier for a backend running on a specific segment.
+ */
+typedef struct
+{
+ int32 segid;
+ int32 pid;
+} gp_segment_pid;
+
+/*
+ * Wire format for the backend-info (CDB segment PIDs) reply.
+ */
+typedef struct
+{
+ int reqid;
+ int length;
+ PGPROC *proc;
+ PG_QS_RequestResult result_code;
+ int number;
+ gp_segment_pid pids[FLEXIBLE_ARRAY_MEMBER];
+} backend_info;
+
+#define BASE_SIZEOF_GP_BACKEND_INFO (offsetof(backend_info, pids))
+
+/*
+ * Parameters passed through shared memory from the requestor to the signal
+ * handler, controlling what the handler should collect and how.
+ */
+typedef struct
+{
+ ProcSignalReason reason;
+ int reqid;
+} pg_qs_params;
+
+/*
+ * Context threaded through the plan-tree walker.
+ * per_node_stats accumulates one GpscNodeSample per visited node.
+ */
+typedef struct QsWalkerContext
+{
+ List *per_node_stats;
+ int32_t parent_plan_node_id;
+ int32_t slice_id; /* slice owning the node being visited */
+ bool finalize; /* true only in pg_qs_executor end */
+ TimestampTz ts_now;
+ int32_t tmid;
+} QsWalkerContext;
+
+/*
+ * Result code for the chunked shm_mq send helper.
+ */
+typedef enum
+{
+ MSG_BY_PARTS_SUCCEEDED,
+ MSG_BY_PARTS_FAILED
+} msg_by_parts_result;
+
+extern bool pg_qs_enable;
+extern bool pg_qs_timing;
+extern bool pg_qs_buffers;
+extern List *QueryDescStack;
+extern pg_qs_params *params;
+extern shm_mq *mq;
+extern uint32 *mq_req_id;
+
+/*
+ * Per-backend trace_id slots, indexed by BackendId (1..MaxBackends; slot 0 for
+ * InvalidBackendId is unused). The single shared `params` cannot carry the
+ * trace across an asynchronous ProcSignal: two concurrent collections would
+ * clobber it and a signaled backend would stamp its batch with the wrong
+ * trace. The dispatcher writes qs_trace_slots[target->backendId] before
+ * signalling; the signaled backend reads qs_trace_slots[MyBackendId]. The slot
+ * is keyed by backend, not by collection, so two overlapping collections of the
+ * same backend still share one slot -- the caller must not poll one pid twice
+ * concurrently.
+ */
+extern char (*qs_trace_slots)[GPSC_TRACE_ID_LEN];
+
+extern ProcSignalReason QueryStatePollReason;
+extern ProcSignalReason BackendInfoPollReason;
+
+/*
+ * pg_qs_init -- register shared memory, custom signals, GUC variables and the
+ * executor hooks. Must be called from _PG_init() during
+ * shared_preload_libraries processing.
+ */
+extern void pg_qs_init(void);
+
+/* Custom signal handlers registered with RegisterCustomProcSignalHandler. */
+extern void SendQueryState(void);
+extern void SendCdbComponents(void);
+
+typedef void (*qs_planstate_walker_callback)(PlanState *, QsWalkerContext *);
+extern QueryDesc *get_toppest_query(void);
+extern void gpsc_reset_node_roll_state(void);
+
+#ifdef __cplusplus
+}
+#endif
+#endif /* __PG_QUERY_STATE_H__ */
diff --git a/gpcontrib/gp_stats_collector/src/pg_query_state/qs_types.h b/gpcontrib/gp_stats_collector/src/pg_query_state/qs_types.h
new file mode 100644
index 00000000000..79312f09acf
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/src/pg_query_state/qs_types.h
@@ -0,0 +1,112 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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.
+ *
+ * qs_types.h
+ * Per-node sample type collected by the pg_query_state plan-tree walker.
+ *
+ * IDENTIFICATION
+ * gpcontrib/gp_stats_collector/src/pg_query_state/qs_types.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef QS_TYPES_H
+#define QS_TYPES_H
+
+#include <stdint.h>
+#include <stdbool.h>
+
+#define GPSC_TRACE_ID_LEN 16
+#define MAX_RELNAME_LEN 64
+
+/*
+ * parent_plan_node_id of a root node. Not 0: plan_node_id counters start at 0
+ * (setrefs.c, and GPORCA's GetNextPlanId), so 0 is always a real node.
+ */
+#define GPSC_NO_PARENT_PLAN_NODE_ID (-1)
+
+/*
+ * Negative segids reported in per-node samples and in the participant list.
+ * Real segments report GpIdentity.segindex, which is >= 0. On the coordinator
+ * host it is -1 for the QD and for the entry-db QE alike, so the entry-db is
+ * re-stamped: consumers key their dedup and their barrier on this value, and
+ * two backends sharing it means one of them is silently dropped.
+ */
+#define GPSC_SEGID_QD (-1)
+#define GPSC_SEGID_ENTRY_DB (-2)
+
+/*
+ * Execution phase of a single plan node as observed at signal time.
+ */
+typedef enum QsNodeStatus
+{
+ QS_NODE_STATUS_UNSPECIFIED = 0,
+ QS_NODE_STATUS_INITIALIZED = 1, /* instrumentation allocated but not yet started */
+ QS_NODE_STATUS_EXECUTING = 2, /* currently inside a tuple-fetch call */
+ QS_NODE_STATUS_FINISHED = 3 /* at least one full loop completed */
+} QsNodeStatus;
+
+typedef struct GpscNodeSample
+{
+ int32_t tmid; /* transaction/time id (gp_gettmid) */
+ int32_t ssid; /* gp_session_id */
+ int32_t ccnt; /* gp_command_count */
+ int32_t plan_node_id; /* Plan.plan_node_id */
+ int32_t parent_plan_node_id; /* parent's plan_node_id, or
+ * GPSC_NO_PARENT_PLAN_NODE_ID at the root */
+ int32_t node_tag; /* nodeTag(plan) */
+ int32_t slice_id; /* currentSliceId */
+ int32_t segindex; /* GpIdentity.segindex */
+ int32_t pid; /* MyProcPid of the sampled backend */
+ int32_t dbid; /* GpIdentity.dbid */
+ int32_t relation_oid; /* OID of scanned relation, or 0 */
+ double plan_rows; /* optimizer row estimate */
+ double ntuples; /* Instrumentation.ntuples */
+ double tuplecount; /* Instrumentation.tuplecount (in-progress loop) */
+ double nloops; /* Instrumentation.nloops */
+ double startup; /* Instrumentation.startup (seconds) */
+ double total; /* Instrumentation.total (seconds) */
+ double firsttuple; /* Instrumentation.firsttuple (seconds) */
+ uint64_t shared_blks_hit;
+ uint64_t shared_blks_read;
+ QsNodeStatus node_status;
+ bool eof; /* Instrumentation.eof: node exhausted for
+ * the current cycle (last fetch returned no
+ * tuple). Lets consumers tell a finished
+ * node from one still actively producing. */
+ /*
+ * Spill, from the GP-specific Instrumentation fields. Reliable once the node
+ * is finalized; a mid-run snapshot is a lower bound (Sort/HashJoin populate
+ * these only at eager-free / explain-end).
+ */
+ bool workfile_created; /* Instrumentation.workfileCreated */
+ int64_t workmem_used; /* Instrumentation.workmemused (bytes) */
+ int64_t workmem_wanted; /* Instrumentation.workmemwanted (bytes); >0 == spilled */
+ /*
+ * Derived rate fields, computed in signal_handler from the per-node rolling
+ * state (previous ntuples and sample time) rather than read from
+ * Instrumentation. Zero on the node's first sample.
+ */
+ double ntuples_delta; /* tuples produced since the previous sample */
+ double tuples_per_sec; /* ntuples_delta divided by the sample interval */
+ double time_since_init_sec; /* seconds since the node's first sample */
+ bool stalled; /* executing but produced no new tuples and not at eof */
+ char relation_name[MAX_RELNAME_LEN];
+} GpscNodeSample;
+
+#endif /* QS_TYPES_H */
diff --git a/gpcontrib/gp_stats_collector/src/pg_query_state/signal_handler.c b/gpcontrib/gp_stats_collector/src/pg_query_state/signal_handler.c
new file mode 100644
index 00000000000..843bbe9a972
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/src/pg_query_state/signal_handler.c
@@ -0,0 +1,1000 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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, BitmapAnd/Or, SubqueryScan, CustomScan,
+ * init-plans, and sub-plans).
+ *
+ * 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_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=%lu shared_blks_read=%lu "
+ "workfile_created=%d workmem_used=%ld workmem_wanted=%ld "
+ "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,
+ s->shared_blks_hit, s->shared_blks_read,
+ (int) s->workfile_created, (long) s->workmem_used, (long) 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 queryDesc is
+ * NULL. Intended for the coordinator (QD) only: on a QE the plan subtree can
+ * reach child PlanStates from other slices that are not instantiated here.
+ */
+static char *
+build_plan_doc(QueryDesc *queryDesc, ExplainFormat format)
+{
+ ExplainState *es;
+
+ if (queryDesc == NULL)
+ return NULL;
+
+ HOLD_INTERRUPTS();
+ {
+ es = NewExplainState();
+ es->format = format;
+ es->verbose = true;
+ es->costs = true;
+ es->runtime = true;
+ ExplainBeginOutput(es);
+ ExplainOpenGroup("Query", NULL, true, es);
+ ExplainPrintPlan(es, queryDesc);
+ ExplainCloseGroup("Query", NULL, true, es);
+ ExplainEndOutput(es);
+ }
+ RESUME_INTERRUPTS();
+
+ return es->str->data;
+}
+
+/*
+ * SendQueryState -- handler for QueryStatePollReason.
+ *
+ * Fired asynchronously when another backend (or the monitoring function)
+ * sends QueryStatePollReason to this process.
+ *
+ * Collects a plan-tree snapshot via runtime_explain(), logs it via
+ * qs_debug_node_stats(), then syncs the emitter config and pushes the whole
+ * snapshot to the yagpcc UDS sink via emit_node_batch(). On the coordinator
+ * it additionally pushes the deparsed plan document (SetQueryPlanReq), which
+ * the compact per-node stats cannot reconstruct; that push is rate-limited to
+ * once per PLAN_DOC_RESEND_INTERVAL_MS per query.
+ *
+ * The entire body runs inside a dedicated MemoryContext that is deleted on
+ * exit, preventing any leaks into the backend's long-lived contexts. Any
+ * errors are swallowed with FlushErrorState() to avoid crashing the backend.
+ */
+void
+SendQueryState(void)
+{
+ int saved_errno = errno;
+ MemoryContext volatile oldcontext = CurrentMemoryContext;
+ MemoryContext volatile qs_context = NULL;
+ QueryDesc *qd;
+
+ if (!pg_qs_enable)
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ if (!list_length(QueryDescStack))
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ if (MyBackendId < 1 || MyBackendId > MaxBackends)
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ if (stack_is_too_deep())
+ {
+ elog(DEBUG1, "pg_query_state: skipping poll, call stack too deep");
+ errno = saved_errno;
+ return;
+ }
+
+ qd = get_toppest_query();
+ if (qd == NULL || qd->planstate == NULL || qd->estate == NULL)
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ HOLD_INTERRUPTS();
+ PG_TRY();
+ {
+ List *qs_result;
+ TimestampTz now = GetCurrentTimestamp();
+
+ qs_context = AllocSetContextCreate(TopMemoryContext,
+ "pg_query_state signal context",
+ ALLOCSET_DEFAULT_SIZES);
+ oldcontext = MemoryContextSwitchTo(qs_context);
+
+ qs_result = runtime_explain(now);
+ qs_debug_node_stats(qs_result);
+ gpsc_qs_sync_config();
+ emit_node_batch(qs_result, qs_trace_slots[MyBackendId]);
+
+ if (Gp_role == GP_ROLE_DISPATCH &&
+ IsTransactionState() && CurrentResourceOwner != NULL)
+ {
+ bool is_same_query;
+ bool is_stale;
+ int32_t tmid;
+
+ gp_gettmid(&tmid);
+ is_same_query = (tmid == last_sent_query_key.tmid &&
+ gp_session_id == last_sent_query_key.ssid &&
+ gp_command_count == last_sent_query_key.ccnt);
+
+ is_stale = !is_same_query ||
+ TimestampDifferenceExceeds(last_sent_query_key.at,
+ now,
+ PLAN_DOC_RESEND_INTERVAL_MS);
+
+ if (is_stale)
+ {
+ char *plan_doc = build_plan_doc(qd, EXPLAIN_FORMAT_JSON);
+
+ gpsc_emit_query_plan(tmid, gp_session_id, gp_command_count,
+ plan_doc, EXPLAIN_FORMAT_JSON);
+
+ last_sent_query_key.tmid = tmid;
+ last_sent_query_key.ssid = gp_session_id;
+ last_sent_query_key.ccnt = gp_command_count;
+ last_sent_query_key.at = now;
+ }
+ }
+ }
+ PG_CATCH();
+ {
+ MemoryContextSwitchTo(oldcontext);
+
+ if (!elog_dismiss(WARNING))
+ {
+ if (qs_context)
+ MemoryContextDelete(qs_context);
+ RESUME_INTERRUPTS();
+ errno = saved_errno;
+ PG_RE_THROW();
+ }
+ }
+ PG_END_TRY();
+
+ MemoryContextSwitchTo(oldcontext);
+ if (qs_context)
+ MemoryContextDelete(qs_context);
+ RESUME_INTERRUPTS();
+ errno = saved_errno;
+}
+
+/*
+ * fill_segpid -- append (segid, pid) pairs from one CDB segment's activelist.
+ *
+ * msg->pids[] has room for exactly 'cap' entries in total (not 'cap' more).
+ * *index is the running write position, shared across all calls for one
+ * message; it is advanced past every entry actually written.
+ *
+ * Entries are skipped when the descriptor has no live backend pid yet, so the
+ * final *index may be LESS than the capacity estimated by the caller. The
+ * caller must derive both msg->number and msg->length from the final *index,
+ * never from the estimate.
+ *
+ * `is_entry_db` selects which of cdbs->{segment_db_info,entry_db_info} the
+ * caller is walking. An entry-db descriptor carries segindex -1, the same
+ * value the QD itself reports, so its entries go out as GPSC_SEGID_ENTRY_DB.
+ *
+ * Returns true if the capacity was hit and one or more writable entries were
+ * dropped.
+ */
+static bool
+fill_segpid(CdbComponentDatabaseInfo *segInfo, backend_info *msg, Size cap,
+ Size *index, bool is_entry_db)
+{
+ ListCell *lc;
+ gp_segment_pid *segpid;
+ SegmentDatabaseDescriptor *dbdesc;
+
+ foreach(lc, segInfo->activelist)
+ {
+ dbdesc = (SegmentDatabaseDescriptor *) lfirst(lc);
+ if (!dbdesc || dbdesc->backendPid <= 0)
+ continue;
+
+ if (!is_entry_db && dbdesc->segindex < 0)
+ continue;
+
+ if (*index >= cap)
+ return true;
+
+ segpid = &msg->pids[(*index)++];
+ segpid->pid = dbdesc->backendPid;
+ segpid->segid = is_entry_db ? GPSC_SEGID_ENTRY_DB : dbdesc->segindex;
+ }
+
+ return false;
+}
+
+static int
+count_active(CdbComponentDatabaseInfo *dbs, Size n)
+{
+ int cnt = 0;
+ for (Size i = 0; i < n; ++i)
+ {
+ cnt += list_length(dbs[i].activelist);
+ }
+
+ return cnt;
+}
+
+/*
+ * SendCdbComponents -- handler for BackendInfoPollReason (QD only).
+ *
+ * Collects the list of active QE (segid, pid) pairs from the CDB component
+ * database and sends them back to the requestor through shm_mq as a
+ * backend_info message.
+ *
+ * Side effects:
+ * - Calls cdbcomponent_getCdbComponents(); the returned structure is owned
+ * and cached by the CDB component cache (CdbComponentsContext), NOT by
+ * the local context below, and must not be freed here.
+ * - Only the locally built backend_info message is allocated in the
+ * short-lived context, which is deleted on every exit path.
+ */
+void
+SendCdbComponents(void)
+{
+ int saved_errno = errno;
+ shm_mq_handle *volatile mqh = NULL;
+ CdbComponentDatabases *cdbs;
+ MemoryContext volatile oldctx = CurrentMemoryContext;
+ MemoryContext volatile ctx = NULL;
+ Size index = 0;
+ msg_by_parts_result send_result = MSG_BY_PARTS_SUCCEEDED;
+
+ if (!mq || shm_mq_get_sender(mq) != MyProc || !mq_req_id)
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ if (!params || params->reason != BackendInfoPollReason)
+ {
+ errno = saved_errno;
+ return;
+ }
+
+ HOLD_INTERRUPTS();
+ PG_TRY();
+ {
+ ctx = AllocSetContextCreate(TopMemoryContext,
+ "pg_query_state SendCdbComponents", ALLOCSET_DEFAULT_SIZES);
+ oldctx = MemoryContextSwitchTo(ctx);
+
+ mqh = shm_mq_attach(mq, NULL, NULL);
+
+ if (Gp_role != GP_ROLE_DISPATCH)
+ {
+ elog(DEBUG1, "pg_query_state: SendCdbComponents: running not on QD");
+ shm_mq_msg error_msg = {*mq_req_id, BASE_SIZEOF_SHM_MQ_MSG,
+ MyProc, WRONG_ROLE};
+ send_result = send_msg_by_parts(mqh, error_msg.length, &error_msg);
+ }
+ else if (!pg_qs_enable)
+ {
+ elog(DEBUG1, "pg_query_state: SendCdbComponents: module disabled");
+ shm_mq_msg disabled_msg = {*mq_req_id, BASE_SIZEOF_SHM_MQ_MSG,
+ MyProc, STAT_DISABLED};
+ send_result = send_msg_by_parts(mqh, disabled_msg.length, &disabled_msg);
+ }
+ else if (list_length(QueryDescStack) == 0)
+ {
+ elog(DEBUG1, "pg_query_state: SendCdbComponents: no active query");
+ shm_mq_msg not_running_msg = {*mq_req_id, BASE_SIZEOF_SHM_MQ_MSG,
+ MyProc, QUERY_NOT_RUNNING};
+ send_result = send_msg_by_parts(mqh, not_running_msg.length, ¬_running_msg);
+ }
+ else
+ {
+ MemoryContextSwitchTo(oldctx);
+ cdbs = cdbcomponent_getCdbComponents();
+ MemoryContextSwitchTo(ctx);
+
+ int qecount = count_active(cdbs->entry_db_info, cdbs->total_entry_dbs)
+ + count_active(cdbs->segment_db_info, cdbs->total_segment_dbs);
+
+ size_t bufsz = BASE_SIZEOF_GP_BACKEND_INFO + sizeof(gp_segment_pid) * qecount;
+ backend_info *msg = (backend_info *) palloc0(bufsz);
+
+ bool truncated = false;
+
+ for (int i = 0; i < cdbs->total_segment_dbs; ++i)
+ {
+ CdbComponentDatabaseInfo *segInfo = &cdbs->segment_db_info[i];
+ truncated |= fill_segpid(segInfo, msg, qecount, &index, false);
+ }
+
+ for (int i = 0; i < cdbs->total_entry_dbs; ++i)
+ {
+ CdbComponentDatabaseInfo *segInfo = &cdbs->entry_db_info[i];
+ truncated |= fill_segpid(segInfo, msg, qecount, &index, true);
+ }
+
+ if (truncated)
+ {
+ elog(WARNING, "pg_query_state: SendCdbComponents: backend list truncated at %d of %d entries",
+ (int) index, qecount);
+ }
+
+ msg->reqid = *mq_req_id;
+ msg->length = BASE_SIZEOF_GP_BACKEND_INFO + sizeof(gp_segment_pid) * index;
+ msg->result_code = QS_RETURNED;
+ Assert(index <= qecount);
+ msg->number = index;
+ send_result = send_msg_by_parts(mqh, msg->length, msg);
+ }
+
+ if (send_result != MSG_BY_PARTS_SUCCEEDED)
+ {
+ elog(DEBUG1, "pg_query_state: SendCdbComponents: send failed (%d)",
+ (int) send_result);
+ }
+
+ shm_mq_detach(mqh);
+ mqh = NULL;
+ }
+ PG_CATCH();
+ {
+ if (mqh)
+ {
+ shm_mq_detach(mqh);
+ mqh = NULL;
+ }
+ MemoryContextSwitchTo(oldctx);
+
+ if (!elog_dismiss(WARNING))
+ {
+ if (ctx)
+ MemoryContextDelete(ctx);
+
+ RESUME_INTERRUPTS();
+ errno = saved_errno;
+ PG_RE_THROW();
+ }
+ }
+ PG_END_TRY();
+
+ MemoryContextSwitchTo(oldctx);
+ if (ctx)
+ MemoryContextDelete(ctx);
+
+ RESUME_INTERRUPTS();
+ errno = saved_errno;
+}
diff --git a/gpcontrib/gp_stats_collector/test/Makefile b/gpcontrib/gp_stats_collector/test/Makefile
new file mode 100644
index 00000000000..3931f4d9592
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/Makefile
@@ -0,0 +1,22 @@
+# Regression tests for the gp_stats_collector pg_query_state signal API.
+#
+# Self-contained installcheck suite: the extension must already be built and
+# installed (make -C .. install) and loaded via shared_preload_libraries in the
+# target cluster. Run with:
+#
+# make -C gpcontrib/gp_stats_collector/test installcheck
+#
+# pg_regress defaults to ./sql/<name>.sql and ./expected/<name>.out.
+
+REGRESS = gpsc_pg_query_state
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = gpcontrib/gp_stats_collector/test
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/gpcontrib/gp_stats_collector/test/crash/README.md b/gpcontrib/gp_stats_collector/test/crash/README.md
new file mode 100644
index 00000000000..473e8300541
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/crash/README.md
@@ -0,0 +1,128 @@
+<!--
+ 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.
+-->
+
+# gp_stats_collector crash test
+
+Liveness test: with the runtime query-state feature fully enabled and a poller
+tracing every running query, Cloudberry must not crash and queries must still
+finish with the same results as without the feature.
+
+Driven by `.github/workflows/gpsc-crash-test.yaml`, which runs on every push and
+pull request against `REL_2_STABLE`, and on `workflow_dispatch`. It is not path
+filtered on purpose: the feature walks a *live* plan tree and calls runtime
+`EXPLAIN` on it, so a change anywhere in the executor — a new plan node, a
+different `PlanState` lifecycle, altered `Instrumentation` timing — can break it,
+not just a change under `gpcontrib/gp_stats_collector/`.
+
+One build, one demo cluster, two `installcheck-parallel` passes on it:
+
+1. **baseline** — feature OFF (stock Cloudberry, module not preloaded) → record
+ failed tests.
+2. **traced** — feature ON + poller running → record failed tests, then the
+ crash gate.
+
+Between the two passes the job drops the `regression` database and every
+`regress*`/`mdb*` role. `installcheck` recreates its database each pass, but
+`CREATE ROLE` makes cluster-global roles that outlive it, so without this the
+traced pass would fail in `test_setup` with "role already exists".
+
+## The verdict
+
+**Hard verdict: the crash gate** (no PANIC / signal / segment down / dead
+coordinator). The failed-test delta `traced \ baseline` is reported for
+information only and does **not** fail the job: `installcheck-parallel` is not
+diff-deterministic, so a tracing-only failure is not, by itself, a regression —
+inspect the uploaded `run2-traced.diffs` by hand.
+
+One test is carved out of that delta as known-flaky: `strings`. Its QD
+parse-time warnings (`nonstandard use of \\`, from `scan.l`'s
+`escape_string_warning`) re-emit non-deterministically when the poller's
+`ProcSignal` lands mid-statement. The query has no runtime stats to report, so
+the diff is client-message noise rather than a correctness signal.
+
+Workload is `installcheck-parallel` (upstream `parallel_schedule`): fast and
+fault-free. Because it injects no faults, any PANIC in the logs is a genuine
+crash, which keeps the crash gate simple and honest.
+
+## Running it locally
+
+Build the tree and create a demo cluster as usual, then from the source root:
+
+```bash
+source /usr/local/cloudberry-db/cloudberry-env.sh
+source gpAux/gpdemo/gpdemo-env.sh
+CRASH=gpcontrib/gp_stats_collector/test/crash
+
+# 1. baseline pass -- a failing installcheck is expected, it is not the verdict
+make -C src/test/regress installcheck-parallel > /tmp/run1-baseline.log 2>&1 || true
+$CRASH/extract_failures.sh /tmp/run1-baseline.log > /tmp/baseline-failures.txt
+
+# 2. clear the roles the baseline leaked
+psql -X -d postgres -c 'DROP DATABASE IF EXISTS regression;'
+psql -X -q -A -t -d postgres \
+ -c "SELECT format('DROP ROLE IF EXISTS %I;', rolname) FROM pg_roles WHERE rolname ~ '^(regress|mdb)'" \
+ | psql -X -d postgres -f -
+
+# 3. turn the feature on -- two restarts: the module has to be loaded before
+# its own GUCs are recognised by gpconfig
+gpconfig -c shared_preload_libraries -v 'gp_stats_collector'
+gpstop -ar
+for guc in pg_query_state.enable pg_query_state.enable_timing \
+ pg_query_state.enable_buffers gpsc.enable gpsc.enable_analyze \
+ gpsc.enable_cdbstats gpsc.report_nested_queries; do
+ gpconfig -c $guc -v on
+done
+gpconfig -c gpsc.logging_mode -v UDS
+gpconfig -c gpsc.uds_path -v /tmp/gpsc_agent.sock
+gpconfig -c compute_query_id -v regress
+gpstop -ar
+psql -X -d postgres -c 'CREATE EXTENSION IF NOT EXISTS gp_stats_collector;'
+
+# 4. sink + tracer
+$CRASH/uds_drain.py --path /tmp/gpsc_agent.sock &
+rm -f /tmp/gpsc_poller.stop
+$CRASH/poller.py --stop-file /tmp/gpsc_poller.stop &
+
+# 5. traced pass
+make -C src/test/regress installcheck-parallel > /tmp/run2-traced.log 2>&1 || true
+touch /tmp/gpsc_poller.stop
+
+# 6. the verdict
+$CRASH/crash_scan.sh gpAux/gpdemo/datadirs
+```
+
+Step 3 is what "feature ON" means; skipping any of it makes the traced pass
+weaker than CI's. If `gpsc.pg_query_state` is not resolvable after step 3 the
+run is vacuous — every poll just errors on a missing function — so CI asserts
+`'gpsc.pg_query_state(int,bytea)'::regprocedure` resolves before starting.
+
+## Files
+
+- `poller.py` — single-process tracer: loops over active client backends in
+ `pg_stat_activity` and calls `gpsc.pg_query_state(pid, trace_id)` on each, with
+ a per-pid cooldown so no pid is polled while a prior poll is in flight (the
+ extension does not support overlapping polls of one pid). Uses `psql`, no
+ Python DB driver. Runs until `--stop-file` appears.
+- `uds_drain.py` — minimal `AF_UNIX` sink for `gpsc.uds_path`; reads and discards
+ so the serialize+send path runs without the real collector agent.
+- `extract_failures.sh` — pulls the sorted set of `... FAILED` test names from a
+ `make installcheck-parallel` log.
+- `crash_scan.sh <log-root>` — the crash gate: log crash markers, `gpstate -e`,
+ `SELECT 1`. Plain `FATAL` is ignored on purpose, being routine during
+ regression runs.
diff --git a/gpcontrib/gp_stats_collector/test/crash/crash_scan.sh b/gpcontrib/gp_stats_collector/test/crash/crash_scan.sh
new file mode 100755
index 00000000000..9b89f904e82
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/crash/crash_scan.sh
@@ -0,0 +1,84 @@
+#!/bin/bash
+# --------------------------------------------------------------------
+#
+# 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.
+#
+# --------------------------------------------------------------------
+# crash_scan.sh <log-root>
+#
+# The crash gate for the tracer run. Independent of test diffs: it decides
+# whether Cloudberry survived the tracing. Exits non-zero (and prints why) on
+# any of:
+# - a crash marker in a coordinator/segment log under <log-root>;
+# - a segment reported down / resyncing by gpstate -e;
+# - the coordinator failing to answer a trivial query.
+#
+# The demo cluster env (gpdemo-env.sh) must be sourced before calling.
+# --------------------------------------------------------------------
+set -uo pipefail
+
+log_root="${1:?usage: crash_scan.sh <log-root>}"
+status=0
+
+# Unambiguous crash markers only. The workload (installcheck-parallel) injects
+# no faults, so a PANIC / signal here is a genuine crash, not a fault-injection
+# recovery test. Excluded on purpose:
+# - plain FATAL: routine during regression (missing role, duplicate object).
+# - "server closed the connection unexpectedly" / "the database system is in
+# recovery mode": routine mirror/walreceiver churn on every restart
+# (gpstop -ar), not a crash.
+# Real crashes are caught here (PANIC, postmaster-wide crash restart, a process
+# killed by a signal) and corroborated by gpstate -e + SELECT 1 below.
+patterns='PANIC|terminating connection because of crash of another server process|was terminated by signal [0-9]'
+
+echo "== crash_scan: log markers under ${log_root} =="
+if hits=$(grep -rERn "${patterns}" "${log_root}" 2>/dev/null); then
+ if [ -n "${hits}" ]; then
+ echo "CRASH: crash markers found:"
+ echo "${hits}" | head -50
+ status=1
+ fi
+fi
+[ "${status}" -eq 0 ] && echo " none"
+
+echo "== crash_scan: segment health (gpstate -e) =="
+if command -v gpstate >/dev/null 2>&1; then
+ gpstate_out=$(gpstate -e 2>&1 || true)
+ echo "${gpstate_out}" | tail -30
+ if echo "${gpstate_out}" | grep -Eiq 'down|resynchroniz|not synchronized|Unsynchronized'; then
+ echo "CRASH: gpstate reports segments down / resyncing"
+ status=1
+ fi
+else
+ echo " gpstate not on PATH (env not sourced?)"
+ status=1
+fi
+
+echo "== crash_scan: coordinator responsive? =="
+if echo 'SELECT 1;' | psql -X -q -A -t -d postgres >/dev/null 2>&1; then
+ echo " SELECT 1 ok"
+else
+ echo "CRASH: coordinator did not answer SELECT 1"
+ status=1
+fi
+
+if [ "${status}" -eq 0 ]; then
+ echo "== crash_scan: PASS (cluster healthy) =="
+else
+ echo "== crash_scan: FAIL (see markers above) =="
+fi
+exit "${status}"
diff --git a/gpcontrib/gp_stats_collector/test/crash/extract_failures.sh b/gpcontrib/gp_stats_collector/test/crash/extract_failures.sh
new file mode 100755
index 00000000000..2a22fe21363
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/crash/extract_failures.sh
@@ -0,0 +1,37 @@
+#!/bin/bash
+# --------------------------------------------------------------------
+#
+# 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.
+#
+# --------------------------------------------------------------------
+# extract_failures.sh <make-log>
+#
+# Prints the sorted, unique set of test names that pg_regress / isolation2
+# reported as FAILED in an installcheck-world make log, one per line. Lines
+# look like "test foo ... FAILED" or " foo ... FAILED"; the test name is
+# the token immediately before the "..." separator.
+# --------------------------------------------------------------------
+set -euo pipefail
+
+log="${1:?usage: extract_failures.sh <make-log>}"
+
+awk '
+ /\.\.\.[[:space:]]*FAILED/ {
+ for (i = 1; i <= NF; i++)
+ if ($i == "...") { print $(i - 1); break }
+ }
+' "${log}" | sort -u
diff --git a/gpcontrib/gp_stats_collector/test/crash/poller.py b/gpcontrib/gp_stats_collector/test/crash/poller.py
new file mode 100755
index 00000000000..f21b234d103
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/crash/poller.py
@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+#
+# 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.
+#
+# poller.py -- crash-test tracer for gp_stats_collector.
+#
+# A single process that, in a loop, finds every active client backend in
+# pg_stat_activity and calls gpsc.pg_query_state(pid, trace_id) against it --
+# "trace everything that moves" -- while a heavy test suite runs concurrently.
+#
+# Deliberately single-process: the extension does not support overlapping polls
+# of the same pid, so no pid is ever polled from two sessions at once. A
+# per-pid cooldown keeps a gap larger than a collection's latency, so even this
+# one session never re-fires a pid whose previous poll may still be in flight.
+#
+# Errors from a poll (backend gone, permission gate, races) are swallowed on
+# purpose: this test cares about cluster liveness, not trace correctness.
+#
+# Stops when the --stop-file appears. Talks to the server through psql, so it
+# needs no Python database driver; the gpdemo environment must be sourced first
+# (PGPORT etc.).
+
+import argparse
+import os
+import secrets
+import subprocess
+import sys
+import time
+
+APP_NAME = "gpsc_crash_poller"
+
+# Backends carrying this application_name are our own psql calls; never trace
+# them, or the poller would chase its own tail.
+ACTIVE_PIDS_SQL = (
+ "SELECT pid FROM pg_stat_activity "
+ "WHERE state = 'active' "
+ "AND backend_type = 'client backend' "
+ "AND coalesce(application_name, '') <> '{app}' "
+ "AND pid <> pg_backend_pid();"
+).format(app=APP_NAME)
+
+
+def psql(dbname, sql, timeout):
+ """Run one SQL statement through psql; return (rc, stdout). Never raises."""
+ env = dict(os.environ, PGAPPNAME=APP_NAME)
+ try:
+ proc = subprocess.run(
+ ["psql", "-X", "-q", "-A", "-t", "-d", dbname, "-c", sql],
+ env=env,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ timeout=timeout,
+ text=True,
+ )
+ return proc.returncode, proc.stdout
+ except subprocess.TimeoutExpired:
+ return -1, "psql timeout"
+ except Exception as exc: # noqa: BLE001 -- liveness test, swallow everything
+ return -1, str(exc)
+
+
+def main():
+ ap = argparse.ArgumentParser(description="gp_stats_collector crash-test poller")
+ ap.add_argument("--dbname", default="postgres",
+ help="stable DB to read pg_stat_activity from (default: postgres)")
+ ap.add_argument("--stop-file", required=True,
+ help="poller exits once this path exists")
+ ap.add_argument("--cooldown", type=float, default=1.0,
+ help="min seconds between two polls of the same pid")
+ ap.add_argument("--round-sleep", type=float, default=0.05,
+ help="seconds to sleep between scan rounds")
+ ap.add_argument("--call-timeout", type=float, default=10.0,
+ help="per-psql-call timeout in seconds")
+ ap.add_argument("--log-every", type=int, default=100,
+ help="print a heartbeat every N rounds")
+ args = ap.parse_args()
+
+ last_polled = {}
+ rounds = 0
+ polls = 0
+ errors = 0
+ logged_error = False # print the first poll error body once, for diagnosis
+ started = time.monotonic()
+
+ print("poller: start (app_name={}, cooldown={}s)".format(APP_NAME, args.cooldown),
+ flush=True)
+
+ while not os.path.exists(args.stop_file):
+ rounds += 1
+ rc, out = psql(args.dbname, ACTIVE_PIDS_SQL, args.call_timeout)
+ if rc != 0:
+ # The coordinator may be momentarily busy/restarting a session; the
+ # crash gate, not the poller, decides whether that is fatal.
+ errors += 1
+ time.sleep(args.round_sleep)
+ continue
+
+ now = time.monotonic()
+ for line in out.splitlines():
+ pid = line.strip()
+ if not pid:
+ continue
+ if now - last_polled.get(pid, 0.0) < args.cooldown:
+ continue
+ trace_hex = secrets.token_hex(16) # exactly 16 bytes -> bytea
+ sql = ("SELECT gpsc.pg_query_state({pid}, '\\x{tid}'::bytea);"
+ .format(pid=pid, tid=trace_hex))
+ prc, pout = psql(args.dbname, sql, args.call_timeout)
+ last_polled[pid] = now
+ polls += 1
+ if prc != 0:
+ errors += 1 # gate/race/backend-gone: expected, not fatal here
+ if not logged_error:
+ # A wall of errors usually means a setup problem (e.g. the
+ # function is missing); surface the first one so the log is
+ # not opaque.
+ print("poller: first poll error: {}".format(pout.strip()),
+ flush=True)
+ logged_error = True
+
+ if rounds % args.log_every == 0:
+ print("poller: rounds={} polls={} errors={} tracked_pids={} elapsed={:.0f}s"
+ .format(rounds, polls, errors, len(last_polled),
+ time.monotonic() - started),
+ flush=True)
+
+ time.sleep(args.round_sleep)
+
+ print("poller: stop (rounds={} polls={} errors={} elapsed={:.0f}s)"
+ .format(rounds, polls, errors, time.monotonic() - started),
+ flush=True)
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/gpcontrib/gp_stats_collector/test/crash/uds_drain.py b/gpcontrib/gp_stats_collector/test/crash/uds_drain.py
new file mode 100755
index 00000000000..39312ebea92
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/crash/uds_drain.py
@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+#
+# 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.
+#
+# uds_drain.py -- minimal Unix-domain-socket sink for gp_stats_collector.
+#
+# gpsc.logging_mode = UDS makes every backend push per-node batches and plan
+# docs to gpsc.uds_path. In the crash test we do not run the real yagpcc; this
+# drain accepts every connection and reads/discards whatever arrives, so the
+# full C++ serialize+send path runs without backpressure drops -- but nothing
+# downstream is exercised or asserted.
+#
+# One drain covers a single-host demo cluster (all segments share the socket
+# path). Runs until killed.
+
+import argparse
+import os
+import socket
+import sys
+import threading
+
+
+def drain_conn(conn):
+ with conn:
+ while True:
+ try:
+ if not conn.recv(65536):
+ return
+ except OSError:
+ return
+
+
+def main():
+ ap = argparse.ArgumentParser(description="gp_stats_collector UDS drain")
+ ap.add_argument("--path", required=True, help="unix socket path to listen on")
+ ap.add_argument("--backlog", type=int, default=128)
+ args = ap.parse_args()
+
+ if os.path.exists(args.path):
+ os.unlink(args.path)
+
+ srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ srv.bind(args.path)
+ srv.listen(args.backlog)
+ os.chmod(args.path, 0o777) # any backend user must be able to connect
+ print("uds_drain: listening on {}".format(args.path), flush=True)
+
+ try:
+ while True:
+ conn, _ = srv.accept()
+ threading.Thread(target=drain_conn, args=(conn,), daemon=True).start()
+ except KeyboardInterrupt:
+ pass
+ finally:
+ srv.close()
+ if os.path.exists(args.path):
+ os.unlink(args.path)
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/gpcontrib/gp_stats_collector/test/expected/gpsc_pg_query_state.out b/gpcontrib/gp_stats_collector/test/expected/gpsc_pg_query_state.out
new file mode 100644
index 00000000000..c3eb3e535d9
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/expected/gpsc_pg_query_state.out
@@ -0,0 +1,57 @@
+-- pg_query_state signal API (extension v1.2): catalog contract + negative paths.
+--
+-- Deterministic coverage only: SQL-visible function/type registration and the
+-- input-validation error branches. The asynchronous happy path (poll a live
+-- query and observe per-node stats) is exercised separately under isolation2,
+-- since it depends on a second running backend and timing.
+-- start_ignore
+CREATE EXTENSION IF NOT EXISTS gp_stats_collector;
+-- end_ignore
+--
+-- Catalog contract: the three SQL-visible functions are registered in the gpsc
+-- schema with the expected return type and dispatch (exec) location.
+-- proexeclocation: c = coordinator, a = any (QE-local), s = all segments.
+--
+SELECT proname,
+ pronargs,
+ prorettype::regtype AS returns,
+ proexeclocation
+FROM pg_proc
+WHERE pronamespace = 'gpsc'::regnamespace
+ AND proname IN ('pg_query_state', 'pg_query_state_backends', 'cbdb_mpp_query_state')
+ORDER BY proname;
+ proname | pronargs | returns | proexeclocation
+-------------------------+----------+---------+-----------------
+ cbdb_mpp_query_state | 2 | void | a
+ pg_query_state | 2 | void | c
+ pg_query_state_backends | 1 | record | c
+(3 rows)
+
+-- Composite identifier type used by the signal layer is present.
+SELECT typname
+FROM pg_type
+WHERE typnamespace = 'gpsc'::regnamespace
+ AND typname = 'gp_segment_pid';
+ typname
+----------------
+ gp_segment_pid
+(1 row)
+
+--
+-- Negative: a backend cannot poll its own state.
+--
+SELECT gpsc.pg_query_state(pg_backend_pid(), '\x00112233445566778899aabbccddeeff'::bytea);
+ERROR: cannot extract state of current process
+SELECT * FROM gpsc.pg_query_state_backends(pg_backend_pid());
+ERROR: cannot extract state of current process
+--
+-- Negative: a pid that maps to no live backend is rejected.
+--
+SELECT gpsc.pg_query_state(-1, '\x00112233445566778899aabbccddeeff'::bytea);
+ERROR: backend with pid=-1 not found
+SELECT * FROM gpsc.pg_query_state_backends(-1);
+ERROR: backend with pid=-1 not found
+-- Cleanup
+-- start_ignore
+DROP EXTENSION gp_stats_collector;
+-- end_ignore
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/.gitignore b/gpcontrib/gp_stats_collector/test/isolation2/.gitignore
new file mode 100644
index 00000000000..0d2848e26fb
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/.gitignore
@@ -0,0 +1,4 @@
+/sql_isolation_testcase.py
+/results/
+/regression.diffs
+/regression.out
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/Makefile b/gpcontrib/gp_stats_collector/test/isolation2/Makefile
new file mode 100644
index 00000000000..abeae91d79a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/Makefile
@@ -0,0 +1,43 @@
+# isolation2 suite for the gp_stats_collector pg_query_state signal API.
+#
+# Multi-session tests that a plain pg_regress run cannot express (one backend
+# polling another). Reuses the core pg_isolation2_regress harness rather than
+# rebuilding it.
+#
+# Prerequisites:
+# - gp_inject_fault available (--enable-faultinjector, on by default) for the
+# happy-path spec.
+# - Extension installed and gp_stats_collector in shared_preload_libraries.
+#
+# The core harness is built on demand by the "harness" target below, so this
+# suite runs from a clean tree with no extra CI step.
+#
+# Run:
+# make -C gpcontrib/gp_stats_collector/test/isolation2 installcheck
+
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+ISO2 = $(top_builddir)/src/test/isolation2
+
+# pg_isolation2_regress is not produced by the gpcontrib build; build it here
+# rather than in CI. Mirrors src/test/isolation2, whose installcheck-* targets
+# all depend on "install".
+harness:
+ $(MAKE) -C $(ISO2) install
+
+# isolation2_main.c hardcodes "python3 ./sql_isolation_testcase.py", resolved
+# from the current directory, so symlink the core driver here before running.
+installcheck: harness
+ @ln -sf $(ISO2)/sql_isolation_testcase.py ./sql_isolation_testcase.py
+ $(ISO2)/pg_isolation2_regress \
+ --init-file=$(top_builddir)/src/test/regress/init_file \
+ --init-file=$(ISO2)/init_file_isolation2 \
+ --inputdir=. --outputdir=. \
+ --bindir='$(bindir)' \
+ --schedule=./isolation2_schedule
+
+clean:
+ rm -rf results/ regression.diffs regression.out sql_isolation_testcase.py
+
+.PHONY: harness installcheck clean
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_backends.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_backends.out
new file mode 100644
index 00000000000..a19477a411b
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_backends.out
@@ -0,0 +1,27 @@
+-- pg_query_state_backends against an *idle* backend returns an empty set.
+--
+-- Deterministic multi-session check with no async race: session 1 tags itself
+-- and sits idle; session 2 looks up its pid and polls it. An idle backend is
+-- "not running a query", so GetRemoteBackendInfo returns QUERY_NOT_RUNNING and
+-- the function yields an empty set (not an error).
+--
+-- Extensions are created by setup.sql.
+
+-- Session 1: tag connection so session 2 can find its pid, then go idle.
+1: SET application_name TO 'qs_idle_target';
+SET
+1: SELECT 1;
+ ?column?
+----------
+ 1
+(1 row)
+
+-- Session 2: idle target -> zero participating backends.
+2: SELECT count(*) AS n_backends FROM gpsc.pg_query_state_backends( (SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_idle_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1));
+ n_backends
+------------
+ 0
+(1 row)
+
+1q: ... <quitting>
+2q: ... <quitting>
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_disabled.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_disabled.out
new file mode 100644
index 00000000000..792610bb011
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_disabled.out
@@ -0,0 +1,61 @@
+-- STAT_DISABLED: when the target backend has pg_query_state.enable = off, its
+-- SendCdbComponents reply is STAT_DISABLED, so polling reports an empty backend
+-- list even though a query is actively running on the segments.
+--
+-- Distinguishes "disabled" from "idle": here the query really is executing
+-- (suspended on a fault), yet the disabled module yields nothing.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_disabled_t (id int) DISTRIBUTED BY (id);
+CREATE
+INSERT INTO qs_disabled_t SELECT generate_series(1, 100);
+INSERT 100
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- Target disables the module for its own session, then runs a query that hangs.
+1: SET application_name TO 'qs_disabled_target';
+SET
+1: SET pg_query_state.enable TO off;
+SET
+1&: SELECT count(*) FROM qs_disabled_t; <waiting ...>
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_wait_until_triggered_fault
+-------------------------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- Running, but module disabled on the target -> empty backend list.
+2: SELECT count(*) AS n_backends FROM gpsc.pg_query_state_backends( (SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_disabled_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1));
+ n_backends
+------------
+ 0
+(1 row)
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+1<: <... completed>
+ count
+-------
+ 100
+(1 row)
+1q: ... <quitting>
+2q: ... <quitting>
+
+DROP TABLE qs_disabled_t;
+DROP
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_perms.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_perms.out
new file mode 100644
index 00000000000..8e8f24e6a41
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_perms.out
@@ -0,0 +1,92 @@
+-- Permission gate: a non-superuser that does not own the target query is
+-- denied; a superuser is allowed.
+--
+-- The gate is superuser() || GetUserId() == proc->roleId. isolation2 runs all
+-- sessions under the same session role (the one that launched the harness), so
+-- the "non-super owner is allowed" branch cannot be expressed here and is not
+-- covered; the deny and superuser-allow branches are.
+--
+-- A non-superuser cannot see another backend's application_name in
+-- pg_stat_activity, so the target pid is captured (as superuser) into a table
+-- before SET ROLE.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_perm_t (id int) DISTRIBUTED BY (id);
+CREATE
+INSERT INTO qs_perm_t SELECT generate_series(1, 100);
+INSERT 100
+CREATE TABLE qs_perm_pid (pid int);
+CREATE
+CREATE ROLE qs_unpriv;
+CREATE
+GRANT SELECT ON qs_perm_pid TO qs_unpriv;
+GRANT
+-- No gpsc grants here on purpose: the extension grants USAGE/EXECUTE to PUBLIC
+-- in its migration, so an ordinary role reaches the roleId gate exactly as it
+-- would in production. This test verifies the gate, not the schema grants.
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+1: SET application_name TO 'qs_perm_target';
+SET
+1&: SELECT count(*) FROM qs_perm_t; <waiting ...>
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_wait_until_triggered_fault
+-------------------------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- Capture the target pid as superuser (sees application_name).
+2: INSERT INTO qs_perm_pid SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_perm_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1;
+INSERT 1
+
+-- Non-superuser, non-owner: both entry points are denied.
+2: SET ROLE qs_unpriv;
+SET
+2: SELECT gpsc.pg_query_state((SELECT pid FROM qs_perm_pid), '\x00112233445566778899aabbccddeeff'::bytea);
+ERROR: permission denied
+2: SELECT * FROM gpsc.pg_query_state_backends((SELECT pid FROM qs_perm_pid));
+ERROR: permission denied
+2: RESET ROLE;
+RESET
+
+-- Superuser: allowed (non-empty backend list).
+2: SELECT count(*) > 0 AS has_backends FROM gpsc.pg_query_state_backends((SELECT pid FROM qs_perm_pid));
+ has_backends
+--------------
+ t
+(1 row)
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+1<: <... completed>
+ count
+-------
+ 100
+(1 row)
+1q: ... <quitting>
+2q: ... <quitting>
+
+DROP OWNED BY qs_unpriv;
+DROP
+DROP ROLE qs_unpriv;
+DROP
+DROP TABLE qs_perm_pid;
+DROP
+DROP TABLE qs_perm_t;
+DROP
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_running.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_running.out
new file mode 100644
index 00000000000..cbced86b14d
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_running.out
@@ -0,0 +1,69 @@
+-- Happy path: a query suspended mid-execution on the QEs is observed live.
+--
+-- Session 1 launches a query that hits an 'executor_pre_tuple_processed'
+-- suspend fault on every primary segment, so its QE backends sit inside the
+-- executor with a live plan tree. Session 2 then:
+-- * pg_query_state_backends(pid) -> at least one participating backend,
+-- * pg_query_state(pid) -> succeeds (fire-and-forget, returns void).
+-- The fault is reset and the suspended query is reaped.
+--
+-- Extensions (gp_stats_collector, gp_inject_fault) come from setup.sql.
+
+CREATE TABLE qs_running_t (id int) DISTRIBUTED BY (id);
+CREATE
+INSERT INTO qs_running_t SELECT generate_series(1, 100);
+INSERT 100
+
+-- Suspend execution on all primary segments.
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- Session 1: tag the connection, then launch a query that hangs on the QEs.
+1: SET application_name TO 'qs_running_target';
+SET
+1&: SELECT count(*) FROM qs_running_t; <waiting ...>
+
+-- Wait until the fault has been hit on the segments.
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_wait_until_triggered_fault
+-------------------------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- Session 2: the running query has live QE backends, and polling succeeds.
+2: SELECT count(*) > 0 AS has_backends FROM gpsc.pg_query_state_backends( (SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_running_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1));
+ has_backends
+--------------
+ t
+(1 row)
+2: SELECT gpsc.pg_query_state( (SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_running_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1), '\x00112233445566778899aabbccddeeff'::bytea);
+ pg_query_state
+----------------
+
+(1 row)
+
+-- Release the fault and reap the suspended query.
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+1<: <... completed>
+ count
+-------
+ 100
+(1 row)
+1q: ... <quitting>
+2q: ... <quitting>
+
+DROP TABLE qs_running_t;
+DROP
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_seg_count.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_seg_count.out
new file mode 100644
index 00000000000..b4ad8a9a84d
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/gpsc_pqs_seg_count.out
@@ -0,0 +1,58 @@
+-- backends reports exactly one participating backend per primary segment for a
+-- single-gang query -- a strict count rather than the has_backends>0 smoke
+-- check in gpsc_pqs_running.
+--
+-- A plain scan+count is one gang, so the QE list must match the number of
+-- primary segments.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_segcount_t (id int) DISTRIBUTED BY (id);
+CREATE
+INSERT INTO qs_segcount_t SELECT generate_series(1, 100);
+INSERT 100
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+1: SET application_name TO 'qs_segcount_target';
+SET
+1&: SELECT count(*) FROM qs_segcount_t; <waiting ...>
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_wait_until_triggered_fault
+-------------------------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+
+-- One backend per primary segment.
+2: SELECT count(*) = (SELECT count(*) FROM gp_segment_configuration WHERE role = 'p' AND content > -1) AS matches_primaries FROM gpsc.pg_query_state_backends( (SELECT pid FROM pg_stat_activity WHERE application_name = 'qs_segcount_target' AND pid <> pg_backend_pid() ORDER BY backend_start LIMIT 1));
+ matches_primaries
+-------------------
+ t
+(1 row)
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid) FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+ gp_inject_fault
+-----------------
+ Success:
+ Success:
+ Success:
+(3 rows)
+1<: <... completed>
+ count
+-------
+ 100
+(1 row)
+1q: ... <quitting>
+2q: ... <quitting>
+
+DROP TABLE qs_segcount_t;
+DROP
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/expected/setup.out b/gpcontrib/gp_stats_collector/test/isolation2/expected/setup.out
new file mode 100644
index 00000000000..f7ab4e44725
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/expected/setup.out
@@ -0,0 +1,6 @@
+-- Shared setup for the pg_query_state isolation2 suite.
+-- pg_isolation2_regress always runs a "setup" test before the schedule.
+CREATE EXTENSION IF NOT EXISTS gp_stats_collector;
+CREATE
+CREATE EXTENSION IF NOT EXISTS gp_inject_fault;
+CREATE
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/isolation2_schedule b/gpcontrib/gp_stats_collector/test/isolation2/isolation2_schedule
new file mode 100644
index 00000000000..5adb0d9cc0a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/isolation2_schedule
@@ -0,0 +1,20 @@
+# pg_query_state isolation2 schedule.
+#
+# gpsc_pqs_backends -- deterministic: an idle backend yields an empty backend
+# list (no fault injector required).
+# gpsc_pqs_running -- happy path: a query suspended mid-execution on the QEs
+# is observed via pg_query_state_backends/pg_query_state.
+# gpsc_pqs_perms -- permission gate: non-superuser non-owner is denied,
+# superuser is allowed.
+# gpsc_pqs_disabled -- STAT_DISABLED: target with pg_query_state.enable=off
+# reports no backends despite a running query.
+# gpsc_pqs_seg_count -- strict count: one participating backend per primary
+# segment for a single-gang query.
+#
+# The gpsc_pqs_* specs after gpsc_pqs_backends use gp_inject_fault (enabled by
+# default) to suspend a running query while it is polled.
+test: gpsc_pqs_backends
+test: gpsc_pqs_running
+test: gpsc_pqs_perms
+test: gpsc_pqs_disabled
+test: gpsc_pqs_seg_count
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_backends.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_backends.sql
new file mode 100644
index 00000000000..4df90380c7a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_backends.sql
@@ -0,0 +1,21 @@
+-- pg_query_state_backends against an *idle* backend returns an empty set.
+--
+-- Deterministic multi-session check with no async race: session 1 tags itself
+-- and sits idle; session 2 looks up its pid and polls it. An idle backend is
+-- "not running a query", so GetRemoteBackendInfo returns QUERY_NOT_RUNNING and
+-- the function yields an empty set (not an error).
+--
+-- Extensions are created by setup.sql.
+
+-- Session 1: tag connection so session 2 can find its pid, then go idle.
+1: SET application_name TO 'qs_idle_target';
+1: SELECT 1;
+
+-- Session 2: idle target -> zero participating backends.
+2: SELECT count(*) AS n_backends FROM gpsc.pg_query_state_backends(
+ (SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_idle_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1));
+
+1q:
+2q:
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_disabled.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_disabled.sql
new file mode 100644
index 00000000000..10bf81d188a
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_disabled.sql
@@ -0,0 +1,36 @@
+-- STAT_DISABLED: when the target backend has pg_query_state.enable = off, its
+-- SendCdbComponents reply is STAT_DISABLED, so polling reports an empty backend
+-- list even though a query is actively running on the segments.
+--
+-- Distinguishes "disabled" from "idle": here the query really is executing
+-- (suspended on a fault), yet the disabled module yields nothing.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_disabled_t (id int) DISTRIBUTED BY (id);
+INSERT INTO qs_disabled_t SELECT generate_series(1, 100);
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- Target disables the module for its own session, then runs a query that hangs.
+1: SET application_name TO 'qs_disabled_target';
+1: SET pg_query_state.enable TO off;
+1&: SELECT count(*) FROM qs_disabled_t;
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- Running, but module disabled on the target -> empty backend list.
+2: SELECT count(*) AS n_backends FROM gpsc.pg_query_state_backends(
+ (SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_disabled_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1));
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+1<:
+1q:
+2q:
+
+DROP TABLE qs_disabled_t;
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_perms.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_perms.sql
new file mode 100644
index 00000000000..eadd5c60b33
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_perms.sql
@@ -0,0 +1,57 @@
+-- Permission gate: a non-superuser that does not own the target query is
+-- denied; a superuser is allowed.
+--
+-- The gate is superuser() || GetUserId() == proc->roleId. isolation2 runs all
+-- sessions under the same session role (the one that launched the harness), so
+-- the "non-super owner is allowed" branch cannot be expressed here and is not
+-- covered; the deny and superuser-allow branches are.
+--
+-- A non-superuser cannot see another backend's application_name in
+-- pg_stat_activity, so the target pid is captured (as superuser) into a table
+-- before SET ROLE.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_perm_t (id int) DISTRIBUTED BY (id);
+INSERT INTO qs_perm_t SELECT generate_series(1, 100);
+CREATE TABLE qs_perm_pid (pid int);
+CREATE ROLE qs_unpriv;
+GRANT SELECT ON qs_perm_pid TO qs_unpriv;
+-- No gpsc grants here on purpose: the extension grants USAGE/EXECUTE to PUBLIC
+-- in its migration, so an ordinary role reaches the roleId gate exactly as it
+-- would in production. This test verifies the gate, not the schema grants.
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+1: SET application_name TO 'qs_perm_target';
+1&: SELECT count(*) FROM qs_perm_t;
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- Capture the target pid as superuser (sees application_name).
+2: INSERT INTO qs_perm_pid SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_perm_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1;
+
+-- Non-superuser, non-owner: both entry points are denied.
+2: SET ROLE qs_unpriv;
+2: SELECT gpsc.pg_query_state((SELECT pid FROM qs_perm_pid), '\x00112233445566778899aabbccddeeff'::bytea);
+2: SELECT * FROM gpsc.pg_query_state_backends((SELECT pid FROM qs_perm_pid));
+2: RESET ROLE;
+
+-- Superuser: allowed (non-empty backend list).
+2: SELECT count(*) > 0 AS has_backends
+ FROM gpsc.pg_query_state_backends((SELECT pid FROM qs_perm_pid));
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+1<:
+1q:
+2q:
+
+DROP OWNED BY qs_unpriv;
+DROP ROLE qs_unpriv;
+DROP TABLE qs_perm_pid;
+DROP TABLE qs_perm_t;
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_running.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_running.sql
new file mode 100644
index 00000000000..3c8386cc247
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_running.sql
@@ -0,0 +1,45 @@
+-- Happy path: a query suspended mid-execution on the QEs is observed live.
+--
+-- Session 1 launches a query that hits an 'executor_pre_tuple_processed'
+-- suspend fault on every primary segment, so its QE backends sit inside the
+-- executor with a live plan tree. Session 2 then:
+-- * pg_query_state_backends(pid) -> at least one participating backend,
+-- * pg_query_state(pid) -> succeeds (fire-and-forget, returns void).
+-- The fault is reset and the suspended query is reaped.
+--
+-- Extensions (gp_stats_collector, gp_inject_fault) come from setup.sql.
+
+CREATE TABLE qs_running_t (id int) DISTRIBUTED BY (id);
+INSERT INTO qs_running_t SELECT generate_series(1, 100);
+
+-- Suspend execution on all primary segments.
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- Session 1: tag the connection, then launch a query that hangs on the QEs.
+1: SET application_name TO 'qs_running_target';
+1&: SELECT count(*) FROM qs_running_t;
+
+-- Wait until the fault has been hit on the segments.
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- Session 2: the running query has live QE backends, and polling succeeds.
+2: SELECT count(*) > 0 AS has_backends FROM gpsc.pg_query_state_backends(
+ (SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_running_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1));
+2: SELECT gpsc.pg_query_state(
+ (SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_running_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1),
+ '\x00112233445566778899aabbccddeeff'::bytea);
+
+-- Release the fault and reap the suspended query.
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+1<:
+1q:
+2q:
+
+DROP TABLE qs_running_t;
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_seg_count.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_seg_count.sql
new file mode 100644
index 00000000000..d887f2e71c1
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/gpsc_pqs_seg_count.sql
@@ -0,0 +1,36 @@
+-- backends reports exactly one participating backend per primary segment for a
+-- single-gang query -- a strict count rather than the has_backends>0 smoke
+-- check in gpsc_pqs_running.
+--
+-- A plain scan+count is one gang, so the QE list must match the number of
+-- primary segments.
+--
+-- Extensions come from setup.sql.
+
+CREATE TABLE qs_segcount_t (id int) DISTRIBUTED BY (id);
+INSERT INTO qs_segcount_t SELECT generate_series(1, 100);
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'suspend', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+1: SET application_name TO 'qs_segcount_target';
+1&: SELECT count(*) FROM qs_segcount_t;
+
+SELECT gp_wait_until_triggered_fault('executor_pre_tuple_processed', 1, dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+
+-- One backend per primary segment.
+2: SELECT count(*) = (SELECT count(*) FROM gp_segment_configuration
+ WHERE role = 'p' AND content > -1) AS matches_primaries
+ FROM gpsc.pg_query_state_backends(
+ (SELECT pid FROM pg_stat_activity
+ WHERE application_name = 'qs_segcount_target' AND pid <> pg_backend_pid()
+ ORDER BY backend_start LIMIT 1));
+
+SELECT gp_inject_fault('executor_pre_tuple_processed', 'reset', dbid)
+ FROM gp_segment_configuration WHERE role = 'p' AND content > -1;
+1<:
+1q:
+2q:
+
+DROP TABLE qs_segcount_t;
diff --git a/gpcontrib/gp_stats_collector/test/isolation2/sql/setup.sql b/gpcontrib/gp_stats_collector/test/isolation2/sql/setup.sql
new file mode 100644
index 00000000000..faec1135517
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/isolation2/sql/setup.sql
@@ -0,0 +1,4 @@
+-- Shared setup for the pg_query_state isolation2 suite.
+-- pg_isolation2_regress always runs a "setup" test before the schedule.
+CREATE EXTENSION IF NOT EXISTS gp_stats_collector;
+CREATE EXTENSION IF NOT EXISTS gp_inject_fault;
diff --git a/gpcontrib/gp_stats_collector/test/sql/gpsc_pg_query_state.sql b/gpcontrib/gp_stats_collector/test/sql/gpsc_pg_query_state.sql
new file mode 100644
index 00000000000..eb07c45afc2
--- /dev/null
+++ b/gpcontrib/gp_stats_collector/test/sql/gpsc_pg_query_state.sql
@@ -0,0 +1,46 @@
+-- pg_query_state signal API (extension v1.2): catalog contract + negative paths.
+--
+-- Deterministic coverage only: SQL-visible function/type registration and the
+-- input-validation error branches. The asynchronous happy path (poll a live
+-- query and observe per-node stats) is exercised separately under isolation2,
+-- since it depends on a second running backend and timing.
+-- start_ignore
+CREATE EXTENSION IF NOT EXISTS gp_stats_collector;
+-- end_ignore
+
+--
+-- Catalog contract: the three SQL-visible functions are registered in the gpsc
+-- schema with the expected return type and dispatch (exec) location.
+-- proexeclocation: c = coordinator, a = any (QE-local), s = all segments.
+--
+SELECT proname,
+ pronargs,
+ prorettype::regtype AS returns,
+ proexeclocation
+FROM pg_proc
+WHERE pronamespace = 'gpsc'::regnamespace
+ AND proname IN ('pg_query_state', 'pg_query_state_backends', 'cbdb_mpp_query_state')
+ORDER BY proname;
+
+-- Composite identifier type used by the signal layer is present.
+SELECT typname
+FROM pg_type
+WHERE typnamespace = 'gpsc'::regnamespace
+ AND typname = 'gp_segment_pid';
+
+--
+-- Negative: a backend cannot poll its own state.
+--
+SELECT gpsc.pg_query_state(pg_backend_pid(), '\x00112233445566778899aabbccddeeff'::bytea);
+SELECT * FROM gpsc.pg_query_state_backends(pg_backend_pid());
+
+--
+-- Negative: a pid that maps to no live backend is rejected.
+--
+SELECT gpsc.pg_query_state(-1, '\x00112233445566778899aabbccddeeff'::bytea);
+SELECT * FROM gpsc.pg_query_state_backends(-1);
+
+-- Cleanup
+-- start_ignore
+DROP EXTENSION gp_stats_collector;
+-- end_ignore
diff --git a/licenses/LICENSE-pg_query_state.txt b/licenses/LICENSE-pg_query_state.txt
new file mode 100644
index 00000000000..d8b1fa5939f
--- /dev/null
+++ b/licenses/LICENSE-pg_query_state.txt
@@ -0,0 +1,18 @@
+Copyright (c) 2016-2025, Postgres Professional
+
+Permission to use, copy, modify, and distribute this software and its
+documentation for any purpose, without fee, and without a written agreement
+is hereby granted, provided that the above copyright notice and this
+paragraph and the following two paragraphs appear in all copies.
+
+IN NO EVENT SHALL POSTGRES PROFESSIONAL BE LIABLE TO ANY PARTY FOR DIRECT,
+INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST
+PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION,
+EVEN IF POSTGRES PROFESSIONAL HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH
+DAMAGE.
+
+POSTGRES PROFESSIONAL SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING,
+BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS IS"
+BASIS, AND POSTGRES PROFESSIONAL HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE,
+SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
diff --git a/pom.xml b/pom.xml
index a3b99aff02c..f91dfea8328 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1275,6 +1275,9 @@ code or new licensing patterns.
<exclude>gpcontrib/gp_stats_collector/gp_stats_collector.control</exclude>
<exclude>gpcontrib/gp_stats_collector/.clang-format</exclude>
<exclude>gpcontrib/gp_stats_collector/Makefile</exclude>
+ <exclude>gpcontrib/gp_stats_collector/test/Makefile</exclude>
+ <exclude>gpcontrib/gp_stats_collector/test/isolation2/Makefile</exclude>
+ <exclude>gpcontrib/gp_stats_collector/test/isolation2/isolation2_schedule</exclude>
<!-- The following files are from yezzey extension,
introduced by Cloudberry.
diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 9cfd09d5f7c..adb600bf5cb 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -1312,15 +1312,37 @@ report_triggers(ResultRelInfo *rInfo, bool show_relname, ExplainState *es)
Instrumentation *instr = rInfo->ri_TrigInstrument + nt;
char *relname;
char *conname = NULL;
+ instr_time starttimespan;
+ double total;
+ double ntuples;
+ double ncalls;
+ if (!es->runtime)
+ {
/* Must clean up instrumentation state */
InstrEndLoop(instr);
+ }
+
+ /* Collect statistic variables */
+ if (!INSTR_TIME_IS_ZERO(instr->starttime))
+ {
+ INSTR_TIME_SET_CURRENT(starttimespan);
+ INSTR_TIME_SUBTRACT(starttimespan, instr->starttime);
+ }
+ else
+ INSTR_TIME_SET_ZERO(starttimespan);
+
+ total = instr->total + INSTR_TIME_GET_DOUBLE(instr->counter)
+ + INSTR_TIME_GET_DOUBLE(starttimespan);
+ ntuples = instr->ntuples + instr->tuplecount;
+ ncalls = ntuples + !INSTR_TIME_IS_ZERO(starttimespan);
+
/*
* We ignore triggers that were never invoked; they likely aren't
* relevant to the current query type.
*/
- if (instr->ntuples == 0)
+ if (ncalls == 0)
continue;
ExplainOpenGroup("Trigger", NULL, true, es);
@@ -1345,10 +1367,10 @@ report_triggers(ResultRelInfo *rInfo, bool show_relname, ExplainState *es)
if (show_relname)
appendStringInfo(es->str, " on %s", relname);
if (es->timing)
- appendStringInfo(es->str, ": time=%.3f calls=%.ld\n",
- 1000.0 * instr->total, instr->ntuples);
+ appendStringInfo(es->str, ": time=%.3f calls=%.0f\n",
+ 1000.0 * total, ncalls);
else
- appendStringInfo(es->str, ": calls=%.ld\n", instr->ntuples);
+ appendStringInfo(es->str, ": calls=%.0f\n", ncalls);
}
else
{
@@ -1357,9 +1379,8 @@ report_triggers(ResultRelInfo *rInfo, bool show_relname, ExplainState *es)
ExplainPropertyText("Constraint Name", conname, es);
ExplainPropertyText("Relation", relname, es);
if (es->timing)
- ExplainPropertyFloat("Time", "ms", 1000.0 * instr->total, 3,
- es);
- ExplainPropertyFloat("Calls", NULL, instr->ntuples, 0, es);
+ ExplainPropertyFloat("Time", "ms", 1000.0 * total, 3, es);
+ ExplainPropertyFloat("Calls", NULL, ncalls, 0, es);
}
if (conname)
@@ -2266,8 +2287,11 @@ ExplainNode(PlanState *planstate, List *ancestors,
* instrumentation results the user didn't ask for. But we do the
* InstrEndLoop call anyway, if possible, to reduce the number of cases
* auto_explain has to contend with.
+ *
+ * If flag es->stateinfo is set, i.e. when printing the current execution
+ * state, this step of cleaning up is missed.
*/
- if (planstate->instrument)
+ if (planstate->instrument && !es->runtime)
InstrEndLoop(planstate->instrument);
/* GPDB_90_MERGE_FIXME: In GPDB, these are printed differently. But does that work
@@ -2304,7 +2328,7 @@ ExplainNode(PlanState *planstate, List *ancestors,
ExplainPropertyFloat("Actual Loops", NULL, nloops, 0, es);
}
}
- else if (es->analyze)
+ else if (es->analyze && !es->runtime)
{
if (es->format == EXPLAIN_FORMAT_TEXT)
appendStringInfoString(es->str, " (never executed)");
@@ -2320,6 +2344,75 @@ ExplainNode(PlanState *planstate, List *ancestors,
}
}
+ /*
+ * Print the progress of node execution at current loop.
+ */
+ if (planstate->instrument && es->analyze && es->runtime)
+ {
+ instr_time starttimespan;
+ double startup_sec;
+ double total_sec;
+ double rows;
+ double loop_num;
+ bool finished;
+
+ if (!INSTR_TIME_IS_ZERO(planstate->instrument->starttime))
+ {
+ INSTR_TIME_SET_CURRENT(starttimespan);
+ INSTR_TIME_SUBTRACT(starttimespan, planstate->instrument->starttime);
+ }
+ else
+ INSTR_TIME_SET_ZERO(starttimespan);
+ startup_sec = 1000.0 * planstate->instrument->firsttuple;
+ total_sec = 1000.0 * (INSTR_TIME_GET_DOUBLE(planstate->instrument->counter)
+ + INSTR_TIME_GET_DOUBLE(starttimespan));
+ rows = planstate->instrument->tuplecount;
+ loop_num = planstate->instrument->nloops + 1;
+
+ finished = planstate->instrument->nloops > 0
+ && !planstate->instrument->running
+ && INSTR_TIME_IS_ZERO(starttimespan);
+
+ if (!finished)
+ {
+ ExplainOpenGroup("Current loop", "Current loop", true, es);
+ if (es->format == EXPLAIN_FORMAT_TEXT)
+ {
+ if (es->timing)
+ {
+ if (planstate->instrument->running)
+ appendStringInfo(es->str,
+ " (Current loop: actual time=%.3f..%.3f rows=%.0f, loop number=%.0f)",
+ startup_sec, total_sec, rows, loop_num);
+ else
+ appendStringInfo(es->str,
+ " (Current loop: running time=%.3f actual rows=0, loop number=%.0f)",
+ total_sec, loop_num);
+ }
+ else
+ appendStringInfo(es->str,
+ " (Current loop: actual rows=%.0f, loop number=%.0f)",
+ rows, loop_num);
+ }
+ else
+ {
+ ExplainPropertyFloat("Actual Loop Number", NULL, loop_num, 0, es);
+ if (es->timing)
+ {
+ if (planstate->instrument->running)
+ {
+ ExplainPropertyFloat("Actual Startup Time", NULL, startup_sec, 3, es);
+ ExplainPropertyFloat("Actual Total Time", NULL, total_sec, 3, es);
+ }
+ else
+ ExplainPropertyFloat("Running Time", NULL, total_sec, 3, es);
+ }
+ ExplainPropertyFloat("Actual Rows", NULL, rows, 0, es);
+ }
+ ExplainCloseGroup("Current loop", "Current loop", true, es);
+ }
+ }
+
/* in text format, first line ends here */
if (es->format == EXPLAIN_FORMAT_TEXT)
appendStringInfoChar(es->str, '\n');
@@ -2874,8 +2967,9 @@ ExplainNode(PlanState *planstate, List *ancestors,
if (es->wal && planstate->instrument)
show_wal_usage(es, &planstate->instrument->walusage);
- /* Prepare per-worker buffer/WAL usage */
- if (es->workers_state && (es->buffers || es->wal) && es->verbose)
+ /* Show worker detail after query execution */
+ if (es->analyze && es->verbose && planstate->worker_instrument
+ && !es->runtime)
{
WorkerInstrumentation *w = planstate->worker_instrument;
@@ -4015,6 +4109,11 @@ show_hash_info(HashState *hashstate, ExplainState *es)
if (hashstate->hinstrument)
memcpy(&hinstrument, hashstate->hinstrument,
sizeof(HashInstrumentation));
+
+ if (hashstate->hashtable)
+ {
+ ExecHashAccumInstrumentation(&hinstrument, hashstate->hashtable);
+ }
/*
* Merge results from workers. In the parallel-oblivious case, the
@@ -4406,21 +4505,16 @@ show_instrumentation_count(const char *qlabel, int which,
if (!es->analyze || !planstate->instrument)
return;
-
+ nloops = planstate->instrument->nloops;
if (which == 2)
- nfiltered = planstate->instrument->nfiltered2;
+ nfiltered = ((nloops > 0) ? planstate->instrument->nfiltered2 / nloops : 0);
else
- nfiltered = planstate->instrument->nfiltered1;
+ nfiltered = ((nloops > 0) ? planstate->instrument->nfiltered1 / nloops : 0);
nloops = planstate->instrument->nloops;
/* In text mode, suppress zero counts; they're not interesting enough */
if (nfiltered > 0 || es->format != EXPLAIN_FORMAT_TEXT)
- {
- if (nloops > 0)
- ExplainPropertyFloat(qlabel, NULL, nfiltered / nloops, 0, es);
- else
- ExplainPropertyFloat(qlabel, NULL, 0.0, 0, es);
- }
+ ExplainPropertyFloat(qlabel, NULL, nfiltered, 0, es);
}
/*
@@ -5078,15 +5172,27 @@ show_modifytable_info(ModifyTableState *mtstate, List *ancestors,
double insert_path;
double other_path;
- InstrEndLoop(outerPlanState(mtstate)->instrument);
+ if (!es->runtime)
+ InstrEndLoop(outerPlanState(mtstate)->instrument);
/* count the number of source rows */
- total = outerPlanState(mtstate)->instrument->ntuples;
other_path = mtstate->ps.instrument->ntuples2;
- insert_path = total - other_path;
- ExplainPropertyFloat("Tuples Inserted", NULL,
- insert_path, 0, es);
+ /*
+ * Insert occurs after extracting row from subplan and in runtime mode
+ * we can appear between these two operations - situation when
+ * total > insert_path + other_path. Therefore we don't know exactly
+ * whether last row from subplan is inserted.
+ * We don't print inserted tuples in runtime mode in order to not print
+ * inconsistent data
+ */
+ if (!es->runtime)
+ {
+ total = outerPlanState(mtstate)->instrument->ntuples;
+ insert_path = total - other_path;
+ ExplainPropertyFloat("Tuples Inserted", NULL, insert_path, 0, es);
+ }
+
ExplainPropertyFloat("Conflicting Tuples", NULL,
other_path, 0, es);
}
diff --git a/src/backend/executor/instrument.c b/src/backend/executor/instrument.c
index 12561e0c051..e85013a4cd5 100644
--- a/src/backend/executor/instrument.c
+++ b/src/backend/executor/instrument.c
@@ -117,6 +117,9 @@ InstrStopNodeSync(Instrumentation *instr, uint64 nTuples)
/* count the returned tuples */
instr->tuplecount += nTuples;
+ /* A zero-tuple stop means the node is exhausted for this cycle. */
+ instr->eof = (nTuples == 0);
+
/* let's update the time only if the timer was requested */
if (instr->need_timer)
{
@@ -207,6 +210,7 @@ InstrEndLoop(Instrumentation *instr)
/* Reset for next cycle (if any) */
instr->running = false;
+ instr->eof = false;
INSTR_TIME_SET_ZERO(instr->starttime);
INSTR_TIME_SET_ZERO(instr->counter);
instr->firsttuple = 0;
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 669b5465d73..fc9694bcc14 100644
--- a/src/backend/storage/ipc/procsignal.c
+++ b/src/backend/storage/ipc/procsignal.c
@@ -99,12 +99,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];
+
static ProcSignalHeader *ProcSignal = NULL;
static ProcSignalSlot *MyProcSignalSlot = NULL;
static bool CheckProcSignal(ProcSignalReason reason);
static void CleanupProcSignalState(int status, Datum arg);
static void ResetProcSignalBarrierBits(uint32 flags);
+static void CheckAndSetCustomSignalInterrupts(void);
static bool ProcessBarrierPlaceholder(void);
/*
@@ -250,6 +258,40 @@ CleanupProcSignalState(int status, Datum arg)
slot->pss_pid = 0;
}
+/* RegisterCustomProcSignalHandler
+ * Assign specific handler of custom process signal with new
+ * ProcSignalReason key.
+ *
+ * This function has to be called in _PG_init function of extensions at the
+ * stage of loading shared preloaded libraries. Otherwise it throws fatal error.
+ *
+ * Return INVALID_PROCSIGNAL if all slots for custom signals are occupied.
+ */
+ProcSignalReason
+RegisterCustomProcSignalHandler(ProcSignalHandler_type handler)
+{
+ ProcSignalReason reason;
+
+
+ if (!process_shared_preload_libraries_in_progress)
+ {
+ ereport(FATAL, (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("cannot register custom signal after startup")));
+ }
+
+ /* Iterate through custom signal slots to find a free one */
+ for (reason = PROCSIG_CUSTOM_1; reason <= PROCSIG_CUSTOM_N; reason++)
+ {
+ if (!CustomInterruptHandlers[reason - PROCSIG_CUSTOM_1])
+ {
+ CustomInterruptHandlers[reason - PROCSIG_CUSTOM_1] = handler;
+ return reason;
+ }
+ }
+
+ return INVALID_PROCSIGNAL;
+}
+
/*
* SendProcSignal
* Send a signal to a Postgres process
@@ -708,7 +750,71 @@ procsignal_sigusr1_handler(SIGNAL_ARGS)
if (CheckProcSignal(PROCSIG_FAILED_LOGIN))
HandleLoginFailed();
+ CheckAndSetCustomSignalInterrupts();
+
SetLatch(MyLatch);
errno = save_errno;
}
+
+/*
+ * Handle receipt of an interrupt indicating any of custom process signals.
+ */
+static void
+CheckAndSetCustomSignalInterrupts()
+{
+ ProcSignalReason reason;
+
+ for (reason = PROCSIG_CUSTOM_1; reason <= PROCSIG_CUSTOM_N; reason++)
+ {
+ if (CheckProcSignal(reason))
+ {
+ /* set interrupt flags */
+ InterruptPending = true;
+ CustomSignalPendings[reason - PROCSIG_CUSTOM_1] = true;
+ }
+ }
+
+ SetLatch(MyLatch);
+}
+
+/*
+ * CheckAndHandleCustomSignals
+ * Check custom signal flags and call handler assigned to that signal
+ * if it is not NULL
+ *
+ * This function is called within CHECK_FOR_INTERRUPTS if interrupt occurred.
+ */
+void
+CheckAndHandleCustomSignals(void)
+{
+ int i;
+
+ /*
+ * This is invoked from ProcessInterrupts(), and since some of the
+ * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
+ * for recursive calls if more signals are received while this runs, so
+ * let's block interrupts until done.
+ */
+ HOLD_INTERRUPTS();
+
+ /* Check on expiring of custom signals and call its handlers if exist */
+ for (i = 0; i < NUM_CUSTOM_PROCSIGNALS; i++)
+ {
+ if (!CustomSignalProcessing[i] && CustomSignalPendings[i])
+ {
+ ProcSignalHandler_type handler;
+
+ CustomSignalPendings[i] = false;
+ handler = CustomInterruptHandlers[i];
+ if (handler != NULL)
+ {
+ CustomSignalProcessing[i] = true;
+ handler();
+ CustomSignalProcessing[i] = false;
+ }
+ }
+ }
+
+ RESUME_INTERRUPTS();
+}
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 828ffe67132..890bb1ecbd4 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -4363,6 +4363,8 @@ ProcessInterrupts(const char* filename, int lineno)
if (ParallelMessagePending)
HandleParallelMessages();
+ CheckAndHandleCustomSignals();
+
if (LogMemoryContextPending)
ProcessLogMemoryContextInterrupt();
}
@@ -4778,7 +4780,7 @@ process_postgres_switches(int argc, char *argv[], GucContext ctx,
* postmaster/postmaster.c (the option sets should not conflict) and with
* the common help() function in main/main.c.
*/
- while ((flag = getopt(argc, argv, "B:bc:C:D:d:EeFf:h:ijk:lMm:N:nOPp:r:R:S:sTt:v:W:-:")) != -1)
+ while ((flag = getopt(argc, argv, "B:bc:C:D:d:EeFf:h:ijk:lMm:N:nOPp:r:R:S:sTt:v:W:Z-:")) != -1)
{
switch (flag)
{
@@ -4946,6 +4948,10 @@ process_postgres_switches(int argc, char *argv[], GucContext ctx,
SetConfigOption("post_auth_delay", optarg, ctx, gucsource);
break;
+ case 'Z':
+ /* ignored for consistency with the postmaster */
+ break;
+
case 'c':
case '-':
{
diff --git a/src/include/commands/explain.h b/src/include/commands/explain.h
index 77cb96f0cab..1078460abd4 100644
--- a/src/include/commands/explain.h
+++ b/src/include/commands/explain.h
@@ -51,6 +51,8 @@ typedef struct ExplainState
bool summary; /* print total planning and execution timing */
bool settings; /* print modified settings */
ExplainFormat format; /* output format */
+ bool runtime; /* print intermediate state of query execution,
+ not after completion */
/* state for output formatting --- not reset for each new plan tree */
int indent; /* current indentation level */
List *grouping_stack; /* format-specific grouping state */
diff --git a/src/include/executor/instrument.h b/src/include/executor/instrument.h
index 4536df3b237..974c315b46e 100644
--- a/src/include/executor/instrument.h
+++ b/src/include/executor/instrument.h
@@ -81,6 +81,9 @@ typedef struct Instrumentation
bool prf_work; /* true if pushdown runtime filters really work */
/* Info about current plan cycle: */
bool running; /* true if we've completed first tuple */
+ bool eof; /* true if the last fetch returned no tuple
+ * (node exhausted for this cycle); safe to read
+ * mid-run, unlike nloops/ntuples */
instr_time starttime; /* Start time of current iteration of node */
instr_time counter; /* Accumulated runtime for this node */
double firsttuple; /* Time for first tuple of this cycle */
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index 0815460c72f..bc9efd6f878 100644
--- a/src/include/storage/procsignal.h
+++ b/src/include/storage/procsignal.h
@@ -15,7 +15,7 @@
#define PROCSIGNAL_H
#include "storage/backendid.h"
-
+#define NUM_CUSTOM_PROCSIGNALS 64
/*
* Reasons for signaling a Postgres child process (a backend or an auxiliary
@@ -29,6 +29,8 @@
*/
typedef enum
{
+ INVALID_PROCSIGNAL = -1, /* Must be first */
+
PROCSIG_CATCHUP_INTERRUPT, /* sinval catchup interrupt */
PROCSIG_NOTIFY_INTERRUPT, /* listen/notify interrupt */
PROCSIG_PARALLEL_MESSAGE, /* message from cooperating parallel backend */
@@ -49,6 +51,14 @@ typedef enum
PROCSIG_FAILED_LOGIN, /* failed login */
+ PROCSIG_CUSTOM_1,
+ /*
+ * PROCSIG_CUSTOM_2,
+ * ...,
+ * PROCSIG_CUSTOM_N-1,
+ */
+ PROCSIG_CUSTOM_N = PROCSIG_CUSTOM_1 + NUM_CUSTOM_PROCSIGNALS - 1,
+
NUM_PROCSIGNALS /* Must be last! */
} ProcSignalReason;
@@ -62,6 +72,9 @@ typedef enum
PROCSIGNAL_BARRIER_PLACEHOLDER = 0
} ProcSignalBarrierType;
+/* Handler of custom process signal */
+typedef void (*ProcSignalHandler_type) (void);
+
/*
* prototypes for functions in procsignal.c
*/
@@ -69,12 +82,15 @@ extern Size ProcSignalShmemSize(void);
extern void ProcSignalShmemInit(void);
extern void ProcSignalInit(int pss_idx);
+extern ProcSignalReason
+RegisterCustomProcSignalHandler(ProcSignalHandler_type handler);
extern int SendProcSignal(pid_t pid, ProcSignalReason reason,
BackendId backendId);
extern uint64 EmitProcSignalBarrier(ProcSignalBarrierType type);
extern void WaitForProcSignalBarrier(uint64 generation);
extern void ProcessProcSignalBarrier(void);
+extern void CheckAndHandleCustomSignals(void);
extern void procsignal_sigusr1_handler(SIGNAL_ARGS);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org