Add views for getting cumulative IC stats

.github/workflows/build-cloudberry.yml

@@ -298,6 +298,7 @@ jobs:
                                "contrib/formatter_fixedwidth:installcheck",
                                "contrib/hstore:installcheck",
                                "contrib/indexscan:installcheck",
+                               "contrib/interconnect:installcheck",
                                "contrib/pg_trgm:installcheck",
                                "contrib/indexscan:installcheck",
                                "contrib/pgcrypto:installcheck",

contrib/interconnect/Makefile

@@ -7,6 +7,10 @@ include $(top_builddir)/contrib/interconnect/Makefile.interconnect
 MODULE_big = interconnect
 PGFILEDESC = "interconnect - inter connection module"
 
+EXTENSION = interconnect
+EXTENSION_VERSION = 1.0
+DATA = interconnect--$(EXTENSION_VERSION).sql
+
 OBJS =  \
 	$(WIN32RES) \
 	ic_common.o \
@@ -33,6 +37,8 @@ OBJS += proxy/ic_proxy_iobuf.o
 SHLIB_LINK += $(filter -luv, $(LIBS))
 endif  # enable_ic_proxy
 
+REGRESS = interconnect
+
 ifdef USE_PGXS
 PG_CONFIG = pg_config
 PGXS := $(shell $(PG_CONFIG) --pgxs)

contrib/interconnect/README.md

@@ -271,3 +271,35 @@ udpifc result:
 
 Notice that: Lower TPS does not mean the protocol is slower, might means that the cpu time taken by the protocol is low. For the udpifc, it satisfies the highest tps required by `cbdb`. at the same time it occupies a lower cpu than other types of interconnect.
 
+# interconnect statistics
+
+This extension provides cumulative interconnect statistics for Cloudberry Database, including queue sizes, buffer usage, retransmits, packet errors, and other UDPIFC‑related metrics. 
+
+It exposes three views with statistics at different aggregation levels:
+gp_interconnect_stats — total cluster‑wide stats;
+gp_interconnect_stats_per_segment — stats per segment;
+gp_interconnect_stats_per_host — stats grouped by host.
+
+## How to create the extension
+
+Add interconnect to shared_preload_libraries and restart the cluster.
+
+```
+gpconfig -c shared_preload_libraries -v \
+  "$(psql -At -c \
+    "SELECT array_to_string( \
+        array_append( \
+          string_to_array( \
+            current_setting('shared_preload_libraries'), \
+            ','), \
+          'interconnect'), \
+        ',')" \
+    postgres)"
+gpstop -ra
+```
+
+Create the extension in your database.
+
+```
+CREATE EXTENSION interconnect;
+```

contrib/interconnect/expected/interconnect.out

@@ -0,0 +1,82 @@
+-- Capture current interconnect stats as baseline for future comparisons
+SELECT * FROM gp_interconnect_stats \gset prev_
+-- Verify that all baseline interconnect statistics are >= 0 (no negative values)
+SELECT
+    :prev_total_recv_queue_size >= 0,
+    :prev_recv_queue_conting_time >= 0,
+    :prev_total_capacity >= 0,
+    :prev_capacity_counting_time >= 0,
+    :prev_total_buffers >= 0,
+    :prev_buffer_counting_time >= 0,
+    :prev_retransmits >= 0,
+    :prev_startup_cached_pkts >= 0,
+    :prev_mismatches >= 0,
+    :prev_crs_errors >= 0,
+    :prev_snd_pkt_num >= 0,
+    :prev_recv_pkt_num >= 0,
+    :prev_disordered_pkt_num >= 0,
+    :prev_duplicate_pkt_num >= 0,
+    :prev_recv_ack_num >= 0,
+    :prev_status_query_msg_num >= 0;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? 
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t
+(1 row)
+
+-- Create test table to generate interconnect traffic
+CREATE TABLE test_ic_data
+AS SELECT generate_series(1, 1000) AS id
+DISTRIBUTED RANDOMLY;
+-- Re-capture current state: overwrite prev with latest values
+SELECT * FROM gp_interconnect_stats \gset prev2_
+-- Check if current statistics are >= baseline values after first data insertion
+SELECT
+    :prev2_total_recv_queue_size >= :prev_total_recv_queue_size,
+    :prev2_recv_queue_conting_time >= :prev_recv_queue_conting_time,
+    :prev2_total_capacity >= :prev_total_capacity,
+    :prev2_capacity_counting_time >= :prev_capacity_counting_time,
+    :prev2_total_buffers >= :prev_total_buffers,
+    :prev2_buffer_counting_time >= :prev_buffer_counting_time,
+    :prev2_retransmits >= :prev_retransmits,
+    :prev2_startup_cached_pkts >= :prev_startup_cached_pkts,
+    :prev2_mismatches >= :prev_mismatches,
+    :prev2_crs_errors >= :prev_crs_errors,
+    :prev2_snd_pkt_num >= :prev_snd_pkt_num,
+    :prev2_recv_pkt_num >= :prev_recv_pkt_num,
+    :prev2_disordered_pkt_num >= :prev_disordered_pkt_num,
+    :prev2_duplicate_pkt_num >= :prev_duplicate_pkt_num,
+    :prev2_recv_ack_num >= :prev_recv_ack_num,
+    :prev2_status_query_msg_num >= :prev_status_query_msg_num;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? 
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t
+(1 row)
+
+-- Insert additional data to further test interconnect statistics changes under load
+INSERT INTO test_ic_data SELECT generate_series(1001, 2000);
+-- Re‑check if current statistics remain >= baseline after second data insertion
+SELECT
+    total_recv_queue_size >= :prev2_total_recv_queue_size,
+    recv_queue_conting_time >= :prev2_recv_queue_conting_time,
+    total_capacity >= :prev2_total_capacity,
+    capacity_counting_time >= :prev2_capacity_counting_time,
+    total_buffers >= :prev2_total_buffers,
+    buffer_counting_time >= :prev2_buffer_counting_time,
+    retransmits >= :prev2_retransmits,
+    startup_cached_pkts >= :prev2_startup_cached_pkts,
+    mismatches >= :prev2_mismatches,
+    crs_errors >= :prev2_crs_errors,
+    snd_pkt_num >= :prev2_snd_pkt_num,
+    recv_pkt_num >= :prev2_recv_pkt_num,
+    disordered_pkt_num >= :prev2_disordered_pkt_num,
+    duplicate_pkt_num >= :prev2_duplicate_pkt_num,
+    recv_ack_num >= :prev2_recv_ack_num,
+    status_query_msg_num >= :prev2_status_query_msg_num
+FROM gp_interconnect_stats;
+ ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? | ?column? 
+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------+----------
+ t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t        | t
+(1 row)
+
+DROP TABLE test_ic_data;
+DROP EXTENSION interconnect;

contrib/interconnect/ic_modules.c

@@ -16,13 +16,16 @@
 #include "ic_common.h"
 #include "tcp/ic_tcp.h"
 #include "udp/ic_udpifc.h"
+#include "storage/ipc.h"
 
 #ifdef ENABLE_IC_PROXY
 #include "proxy/ic_proxy_server.h"
 #endif
 
 PG_MODULE_MAGIC;
 
+shmem_startup_hook_type prev_shmem_startup_hook = NULL;
+
 MotionIPCLayer tcp_ipc_layer = {
     .ic_type = INTERCONNECT_TYPE_TCP,
     .type_name = "tcp",
@@ -141,6 +144,20 @@ MotionIPCLayer udpifc_ipc_layer = {
     .GetMotionSentRecordTypmod = GetMotionSentRecordTypmod,
 };
 
+static shmem_request_hook_type prev_shmem_request_hook = NULL;
+
+static void ic_shmem_request(void);
+
+static void
+InterconnectShmemInit(void)
+{
+    if (prev_shmem_startup_hook)
+        prev_shmem_startup_hook();
+
+    if (Gp_interconnect_type == INTERCONNECT_TYPE_UDPIFC)
+        InterconnectShmemInitUDPIFC();
+}
+
 void
 _PG_init(void)
 {
@@ -153,4 +170,31 @@ _PG_init(void)
 	RegisterIPCLayerImpl(&tcp_ipc_layer);
 	RegisterIPCLayerImpl(&udpifc_ipc_layer);
 	RegisterIPCLayerImpl(&proxy_ipc_layer);
+
+    if (Gp_interconnect_type == INTERCONNECT_TYPE_UDPIFC)
+    {
+	    prev_shmem_request_hook = shmem_request_hook;
+        shmem_request_hook = ic_shmem_request;
+    }
+
+    prev_shmem_startup_hook = shmem_startup_hook;
+    shmem_startup_hook = InterconnectShmemInit;
+}
+
+/*
+ * Requests any additional shared memory required for autoprewarm.
+ */
+static void
+ic_shmem_request(void)
+{
+	if (prev_shmem_request_hook)
+		prev_shmem_request_hook();
+
+	RequestAddinShmemSpace(MAXALIGN(sizeof(ICStatisticsShmem)));
+}
+
+void
+_PG_fini(void)
+{
+    shmem_startup_hook = prev_shmem_startup_hook;
 }

contrib/interconnect/ic_modules.h

@@ -18,5 +18,6 @@ extern MotionIPCLayer proxy_ipc_layer;
 extern MotionIPCLayer udpifc_ipc_layer;
 
 extern void _PG_init(void);
+extern void _PG_fini(void);
 
 #endif // INTER_CONNECT_H

contrib/interconnect/interconnect--1.0.sql

@@ -0,0 +1,110 @@
+/* contrib/interconnect/interconnect--1.0.sql */
+
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION interconnect" to load this file. \quit
+
+CREATE FUNCTION __gp_interconnect_get_stats_f_on_master(
+	OUT gp_segment_id smallint,
+	OUT total_recv_queue_size bigint,
+	OUT recv_queue_conting_time bigint,
+	OUT total_capacity bigint,
+	OUT capacity_counting_time bigint,
+	OUT total_buffers bigint,
+	OUT buffer_counting_time bigint,
+	OUT retransmits bigint,
+	OUT startup_cached_pkts bigint,
+	OUT mismatches bigint,
+	OUT crs_errors bigint,
+	OUT snd_pkt_num bigint,
+	OUT recv_pkt_num bigint,
+	OUT disordered_pkt_num bigint,
+	OUT duplicate_pkt_num bigint,
+	OUT recv_ack_num bigint,
+	OUT status_query_msg_num bigint
+)
+RETURNS SETOF record
+LANGUAGE C VOLATILE EXECUTE ON MASTER
+AS '$libdir/interconnect', 'gp_interconnect_get_stats';
+
+CREATE FUNCTION __gp_interconnect_get_stats_f_on_segments(
+	OUT gp_segment_id smallint,
+	OUT total_recv_queue_size bigint,
+	OUT recv_queue_conting_time bigint,
+	OUT total_capacity bigint,
+	OUT capacity_counting_time bigint,
+	OUT total_buffers bigint,
+	OUT buffer_counting_time bigint,
+	OUT retransmits bigint,
+	OUT startup_cached_pkts bigint,
+	OUT mismatches bigint,
+	OUT crs_errors bigint,
+	OUT snd_pkt_num bigint,
+	OUT recv_pkt_num bigint,
+	OUT disordered_pkt_num bigint,
+	OUT duplicate_pkt_num bigint,
+	OUT recv_ack_num bigint,
+	OUT status_query_msg_num bigint
+)
+RETURNS SETOF record LANGUAGE C VOLATILE EXECUTE ON ALL SEGMENTS
+AS '$libdir/interconnect', 'gp_interconnect_get_stats';
+
+
+-- Cummulative interconnect statistics per segment
+CREATE VIEW gp_interconnect_stats_per_segment AS
+    SELECT c.hostname, s.* FROM (
+        SELECT * FROM __gp_interconnect_get_stats_f_on_master()
+        UNION ALL
+        SELECT * FROM __gp_interconnect_get_stats_f_on_segments()
+    ) s
+    JOIN pg_catalog.gp_segment_configuration AS c
+        ON s.gp_segment_id = c.content AND c.role = 'p';
+
+GRANT SELECT ON gp_interconnect_stats_per_segment TO public;
+
+-- Cummulative interconnect statistics
+CREATE VIEW gp_interconnect_stats AS
+    SELECT
+        sum(total_recv_queue_size) as total_recv_queue_size
+	  , sum(recv_queue_conting_time) as recv_queue_conting_time
+	  , sum(total_capacity) as total_capacity
+	  , sum(capacity_counting_time) as capacity_counting_time
+	  , sum(total_buffers) as total_buffers
+	  , sum(buffer_counting_time) as buffer_counting_time
+	  , sum(retransmits) as retransmits
+	  , sum(startup_cached_pkts) as startup_cached_pkts
+	  , sum(mismatches) as mismatches
+	  , sum(crs_errors) as crs_errors
+	  , sum(snd_pkt_num) as snd_pkt_num
+	  , sum(recv_pkt_num) as recv_pkt_num
+	  , sum(disordered_pkt_num) as disordered_pkt_num
+	  , sum(duplicate_pkt_num) as duplicate_pkt_num
+	  , sum(recv_ack_num) as recv_ack_num
+	  , sum(status_query_msg_num) as status_query_msg_num
+    FROM gp_interconnect_stats_per_segment;
+
+GRANT SELECT ON gp_interconnect_stats TO public;
+
+-- Cummulative interconnect statistics grouped by host
+CREATE VIEW gp_interconnect_stats_per_host AS
+    SELECT
+        hostname
+      , sum(total_recv_queue_size) as total_recv_queue_size
+	  , sum(recv_queue_conting_time) as recv_queue_conting_time
+	  , sum(total_capacity) as total_capacity
+	  , sum(capacity_counting_time) as capacity_counting_time
+	  , sum(total_buffers) as total_buffers
+	  , sum(buffer_counting_time) as buffer_counting_time
+	  , sum(retransmits) as retransmits
+	  , sum(startup_cached_pkts) as startup_cached_pkts
+	  , sum(mismatches) as mismatches
+	  , sum(crs_errors) as crs_errors
+	  , sum(snd_pkt_num) as snd_pkt_num
+	  , sum(recv_pkt_num) as recv_pkt_num
+	  , sum(disordered_pkt_num) as disordered_pkt_num
+	  , sum(duplicate_pkt_num) as duplicate_pkt_num
+	  , sum(recv_ack_num) as recv_ack_num
+	  , sum(status_query_msg_num) as status_query_msg_num
+    FROM gp_interconnect_stats_per_segment
+    GROUP BY hostname;
+
+GRANT SELECT ON gp_interconnect_stats_per_host TO public;

contrib/interconnect/interconnect.control

@@ -0,0 +1,4 @@
+comment = 'Cummulative statistics from UDPIFC interconnect protocol'
+default_version = '1.0'
+relocatable = false
+schema = public