From 84d305c5aba5b333d9eac3d9b4c4eafeae92dc79 Mon Sep 17 00:00:00 2001 From: Jianghua Yang Date: Thu, 3 Sep 2026 23:40:30 +0800 Subject: [PATCH] interconnect: prune cursor IC history table by local xid, not distributed xid SetupUDPIFCInterconnect_Internal() records one CursorICHistoryEntry per interconnect instance on the QD and prunes old entries only when the statement runs in a different transaction than the previous one, using the distributed xid as the transaction identity. Distributed xids are assigned lazily (only for transactions that write or are dispatched two-phase), so autocommit read-only statements always see InvalidDistributedTransactionId on both sides of the comparison. The table is therefore never pruned on read-only sessions and grows by one 48-byte entry per Motion-containing statement for the life of the backend (2.8 MB after 60000 statements, plus the heap fragmentation the steady stream of small long-lived allocations induces). Use MyProc->lxid instead: it is assigned to every top-level transaction, including read-only ones, and stays constant within a transaction block, so entries belonging to cursors that are still open in the current transaction remain protected exactly as before. Verified on a single long-lived session running the queries from #1947: UdpInterconnectMemContext grew 0 -> 2814 KB over 60000 statements before the change and stays at 12 KB after it; RssAnon went 12.7 -> 24.6 MB before and is flat at 14.4 MB after. Pruning still does not happen inside an explicit transaction block or while a cursor is open, and FETCH from an open cursor after 300 intervening statements works. Related: #1947 --- contrib/interconnect/udp/ic_udpifc.c | 32 +++++++++++++++------------- 1 file changed, 17 insertions(+), 15 deletions(-) diff --git a/contrib/interconnect/udp/ic_udpifc.c b/contrib/interconnect/udp/ic_udpifc.c index 0f23ecaad8e..fed89943d31 100644 --- a/contrib/interconnect/udp/ic_udpifc.c +++ b/contrib/interconnect/udp/ic_udpifc.c @@ -46,6 +46,7 @@ #include "nodes/pg_list.h" #include "nodes/print.h" #include "miscadmin.h" +#include "storage/proc.h" #include "libpq/libpq-be.h" #include "port/atomics.h" #include "port/pg_crc32c.h" @@ -353,11 +354,18 @@ struct ReceiveControlInfo CursorICHistoryTable cursorHistoryTable; /* - * Last distributed transaction id when SetupUDPInterconnect is called. - * Coupled with cursorHistoryTable, it is used to handle multiple - * concurrent cursor cases. + * Local (virtual) transaction id of the top-level transaction the last + * time SetupUDPInterconnect was called. Coupled with cursorHistoryTable, + * it is used to handle multiple concurrent cursor cases: entries may only + * be pruned once we are in a different transaction. + * + * We deliberately use MyProc->lxid rather than the distributed xid: a + * distributed xid is only assigned lazily to transactions that write (or + * are dispatched two-phase), so for autocommit read-only statements it is + * always InvalidDistributedTransactionId and the "different transaction" + * test would never fire, letting the history table grow without bound. */ - DistributedTransactionId lastDXatId; + LocalTransactionId lastLxid; }; /* @@ -1894,7 +1902,7 @@ InitMotionUDPIFC(int *listenerSocketFd, int32 *listenerPort) /* allocate a buffer for sending disorder messages */ rx_control_info.disorderBuffer = palloc0(MIN_PACKET_SIZE); - rx_control_info.lastDXatId = InvalidTransactionId; + rx_control_info.lastLxid = InvalidLocalTransactionId; rx_control_info.lastTornIcId = 0; initCursorICHistoryTable(&rx_control_info.cursorHistoryTable); @@ -3573,19 +3581,13 @@ SetupUDPIFCInterconnect_Internal(SliceTable *sliceTable) if (Gp_role == GP_ROLE_DISPATCH) { - DistributedTransactionId distTransId = 0; - TransactionId localTransId = 0; - TransactionId subtransId = 0; - - GetAllTransactionXids(&(distTransId), - &(localTransId), - &(subtransId)); + LocalTransactionId curLxid = MyProc->lxid; /* - * Prune only when we are not in the save transaction and there is a + * Prune only when we are not in the same transaction and there is a * large number of entries in the table */ - if (distTransId != rx_control_info.lastDXatId && rx_control_info.cursorHistoryTable.count > (2 * CURSOR_IC_TABLE_SIZE)) + if (curLxid != rx_control_info.lastLxid && rx_control_info.cursorHistoryTable.count > (2 * CURSOR_IC_TABLE_SIZE)) { if (gp_log_interconnect >= GPVARS_VERBOSITY_DEBUG) elog(DEBUG1, "prune cursor history table (count %d), icid %d", rx_control_info.cursorHistoryTable.count, sliceTable->ic_instance_id); @@ -3595,7 +3597,7 @@ SetupUDPIFCInterconnect_Internal(SliceTable *sliceTable) addCursorIcEntry(&rx_control_info.cursorHistoryTable, sliceTable->ic_instance_id, gp_command_count); /* save the latest transaction id. */ - rx_control_info.lastDXatId = distTransId; + rx_control_info.lastLxid = curLxid; } /* now we'll do some setup for each of our Receiving Motion Nodes. */