Add citus_lock_waits to show locked distributed queries

citus.control

@@ -1,6 +1,6 @@
 # Citus extension
 comment = 'Citus distributed database'
-default_version = '8.0-6'
+default_version = '8.0-7'
 module_pathname = '$libdir/citus'
 relocatable = false
 schema = pg_catalog

src/backend/distributed/Makefile

@@ -17,7 +17,7 @@ EXTVERSIONS = 5.0 5.0-1 5.0-2  \
 	7.3-1 7.3-2 7.3-3 \
 	7.4-1 7.4-2 7.4-3 \
 	7.5-1 7.5-2 7.5-3 7.5-4 7.5-5 7.5-6 7.5-7 \
-	8.0-1 8.0-2 8.0-3 8.0-4 8.0-5 8.0-6
+	8.0-1 8.0-2 8.0-3 8.0-4 8.0-5 8.0-6 8.0-7
 
 # All citus--*.sql files in the source directory
 DATA = $(patsubst $(citus_abs_srcdir)/%.sql,%.sql,$(wildcard $(citus_abs_srcdir)/$(EXTENSION)--*--*.sql))
@@ -227,6 +227,8 @@ $(EXTENSION)--8.0-5.sql: $(EXTENSION)--8.0-4.sql $(EXTENSION)--8.0-4--8.0-5.sql
 	cat $^ > $@
 $(EXTENSION)--8.0-6.sql: $(EXTENSION)--8.0-5.sql $(EXTENSION)--8.0-5--8.0-6.sql
 	cat $^ > $@
+$(EXTENSION)--8.0-7.sql: $(EXTENSION)--8.0-6.sql $(EXTENSION)--8.0-6--8.0-7.sql
+	cat $^ > $@
 
 NO_PGXS = 1
 

src/backend/distributed/citus--8.0-6--8.0-7.sql

@@ -0,0 +1,46 @@
+/* citus--8.0-6--8.0-7 */
+SET search_path = 'pg_catalog';
+
+CREATE VIEW citus.citus_lock_waits AS
+
+WITH 
+citus_dist_stat_activity AS
+(
+  SELECT * FROM citus_dist_stat_activity
+),
+unique_global_wait_edges AS
+(
+	SELECT DISTINCT ON(waiting_node_id, waiting_transaction_num, blocking_node_id, blocking_transaction_num) * FROM dump_global_wait_edges()
+),
+citus_dist_stat_activity_with_node_id AS
+(
+  SELECT
+  citus_dist_stat_activity.*, (CASE citus_dist_stat_activity.master_query_host_name WHEN 'coordinator_host' THEN 0 ELSE pg_dist_node.nodeid END) as initiator_node_id
+  FROM
+  citus_dist_stat_activity LEFT JOIN pg_dist_node
+  ON
+  citus_dist_stat_activity.master_query_host_name = pg_dist_node.nodename AND 
+  citus_dist_stat_activity.master_query_host_port = pg_dist_node.nodeport
+)
+SELECT
+ waiting.pid AS waiting_pid,
+ blocking.pid AS blocking_pid,
+ waiting.query AS blocked_statement,
+ blocking.query AS current_statement_in_blocking_process,
+ waiting.initiator_node_id AS waiting_node_id,
+ blocking.initiator_node_id AS blocking_node_id,
+ waiting.master_query_host_name AS waiting_node_name,
+ blocking.master_query_host_name AS blocking_node_name,
+ waiting.master_query_host_port AS waiting_node_port,
+ blocking.master_query_host_port AS blocking_node_port
+FROM
+ unique_global_wait_edges
+JOIN
+ citus_dist_stat_activity_with_node_id waiting ON (unique_global_wait_edges.waiting_transaction_num = waiting.transaction_number AND unique_global_wait_edges.waiting_node_id = waiting.initiator_node_id)
+JOIN
+ citus_dist_stat_activity_with_node_id blocking ON (unique_global_wait_edges.blocking_transaction_num = blocking.transaction_number AND unique_global_wait_edges.blocking_node_id = blocking.initiator_node_id);
+
+ALTER VIEW citus.citus_lock_waits SET SCHEMA pg_catalog;
+GRANT SELECT ON pg_catalog.citus_lock_waits TO PUBLIC;
+
+RESET search_path;

src/backend/distributed/citus.control

@@ -1,6 +1,6 @@
 # Citus extension
 comment = 'Citus distributed database'
-default_version = '8.0-6'
+default_version = '8.0-7'
 module_pathname = '$libdir/citus'
 relocatable = false
 schema = pg_catalog

src/backend/distributed/test/run_from_same_connection.c

@@ -142,12 +142,12 @@ run_commands_on_session_level_connection_to_node(PG_FUNCTION_ARGS)
 			 "start_session_level_connection_to_node must be called first to open a session level connection");
 	}
 
-	ExecuteCriticalRemoteCommand(connection, queryString);
-
 	appendStringInfo(processStringInfo, ALTER_CURRENT_PROCESS_ID, MyProcPid);
 	appendStringInfo(workerProcessStringInfo, ALTER_CURRENT_WORKER_PROCESS_ID,
 					 GetRemoteProcessId(connection));
 
+	ExecuteCriticalRemoteCommand(connection, queryString);
+
 	/*
 	 * Since we cannot run `ALTER SYSTEM` command within a transaction, we are
 	 * calling it from a self-connected session.