arenadata · Stolb27 · Jul 9, 2021 · Jun 17, 2021 · Jun 24, 2021 · Jun 18, 2021
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
@@ -914,6 +914,45 @@ include_dir 'conf.d'
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-client-connection-check-interval" xreflabel="client_connection_check_interval">
+      <term><varname>client_connection_check_interval</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>client_connection_check_interval</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Sets the time interval between optional checks that the client is still
+        connected, while running queries.  The check is performed by polling
+        the socket, and allows long running queries to be aborted sooner if
+        the kernel reports that the connection is closed.
+       </para>
+       <para>
+        This option is currently available only on systems that support the
+        non-standard <symbol>POLLRDHUP</symbol> extension to the
+        <symbol>poll</symbol> system call, including Linux, or are able to
+        detect client disconnection using POLLHUP option defined by POSIX,
+        including OSX.
+       </para>
+       <para>
+        If the value is specified without units, it is taken as milliseconds.
+        The default value is <literal>0</literal>, which disables connection
+        checks.  Without connection checks, the server will detect the loss of
+        the connection only at the next interaction with the socket, when it
+        waits for, receives or sends data.
+       </para>
+       <para>
+        For the kernel itself to detect lost TCP connections reliably and within
+        a known timeframe in all scenarios including network failure, it may
+        also be necessary to adjust the TCP keepalive settings of the operating
+        system, or the <xref linkend="guc-tcp-keepalives-idle"/>,
+        <xref linkend="guc-tcp-keepalives-interval"/> and
+        <xref linkend="guc-tcp-keepalives-count"/> settings of
+        <productname>PostgreSQL</productname>.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
      </sect2>
      <sect2 id="runtime-config-connection-security">

diff --git a/gpdb-doc/dita/ref_guide/config_params/guc-list.xml b/gpdb-doc/dita/ref_guide/config_params/guc-list.xml
@@ -33,6 +33,9 @@
             <li>
               <xref href="#check_function_bodies"/>
             </li>
+            <li>
+              <xref href="#client_connection_check_interval"/>
+            </li>
             <li>
               <xref href="#client_encoding"/>
             </li>
@@ -895,6 +898,34 @@
       </table>
     </body>
   </topic>
+  <topic id="client_connection_check_interval">
+      <title>client_connection_check_interval</title>
+    <body>
+      <p>Sets the time interval between optional checks that the client is still
+        connected, while running queries. 0 disables connection checks. </p>
+      <table id="client_connection_check_interval_table">
+        <tgroup cols="3">
+          <colspec colnum="1" colname="col1" colwidth="1*"/>
+          <colspec colnum="2" colname="col2" colwidth="1*"/>
+          <colspec colnum="3" colname="col3" colwidth="1*"/>
+          <thead>
+            <row>
+              <entry colname="col1">Value Range</entry>
+              <entry colname="col2">Default</entry>
+              <entry colname="col3">Set Classifications</entry>
+            </row>
+          </thead>
+          <tbody>
+            <row>
+              <entry colname="col1">number of milliseconds</entry>
+              <entry colname="col2">0</entry>
+              <entry colname="col3">master<p>session</p><p>reload</p></entry>
+            </row>
+          </tbody>
+        </tgroup>
+      </table>
+    </body>
+  </topic>
   <topic id="client_encoding">
     <title>client_encoding</title>
     <body>

diff --git a/gpdb-doc/dita/ref_guide/config_params/guc_category-list.xml b/gpdb-doc/dita/ref_guide/config_params/guc_category-list.xml
@@ -75,6 +75,10 @@
         <simpletable id="kh159468" frame="none">
           <strow>
             <stentry>
+              <p>
+                <xref href="guc-list.xml#client_connection_check_interval" type="section"
+                  >client_connection_check_interval</xref>
+              </p>
               <p>
                 <xref href="guc-list.xml#gp_connection_send_timeout" type="section"
                   >gp_connection_send_timeout</xref>

diff --git a/src/backend/access/aocs/aocs_compaction.c b/src/backend/access/aocs/aocs_compaction.c
@@ -31,6 +31,7 @@
 #include "executor/executor.h"
 #include "nodes/execnodes.h"
 #include "storage/lmgr.h"
+#include "utils/faultinjector.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/relcache.h"
@@ -181,7 +182,11 @@ AOCSTruncateToEOF(Relation aorel)
 				segno;
 	LockAcquireResult acquireResult;
 	AOCSFileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	SIMPLE_FAULT_INJECTOR("ao_column_truncate_to_eof");
+#endif
 
 	Assert(RelationIsAoCols(aorel));
 
@@ -241,7 +246,6 @@ AOCSTruncateToEOF(Relation aorel)
 		FreeAllAOCSSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
 
 static void
@@ -436,7 +440,7 @@ AOCSDrop(Relation aorel,
 	int			i,
 				segno;
 	AOCSFileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
 
 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);
 	Assert(RelationIsAoCols(aorel));
@@ -486,7 +490,6 @@ AOCSDrop(Relation aorel,
 		FreeAllAOCSSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
 
 
@@ -516,7 +519,11 @@ AOCSCompact(Relation aorel,
 				segno;
 	LockAcquireResult acquireResult;
 	AOCSFileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	SIMPLE_FAULT_INJECTOR("ao_column_compact");
+#endif
 
 	Assert(RelationIsAoCols(aorel));
 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);
@@ -602,6 +609,4 @@ AOCSCompact(Relation aorel,
 		FreeAllAOCSSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
diff --git a/src/backend/access/aocs/aocsam.c b/src/backend/access/aocs/aocsam.c
@@ -828,7 +828,14 @@ aocs_insert_init(Relation rel, int segno, bool update_mode)
 
 	desc = (AOCSInsertDesc) palloc0(sizeof(AOCSInsertDescData));
 	desc->aoi_rel = rel;
-	desc->appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	desc->appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	if (SIMPLE_FAULT_INJECTOR("ao_column_insert_init_1") == FaultInjectorTypeSkip)
+	{
+		SIMPLE_FAULT_INJECTOR("ao_column_insert_init_2");
+	}
+#endif
 
 	/*
 	 * Writers uses this since they have exclusive access to the lock acquired
@@ -1008,8 +1015,6 @@ aocs_insert_finish(AOCSInsertDesc idesc)
 
 	UpdateAOCSFileSegInfo(idesc);
 
-	UnregisterSnapshot(idesc->appendOnlyMetaDataSnapshot);
-
 	pfree(idesc->fsInfo);
 
 	close_ds_write(idesc->ds, rel->rd_att->natts);

diff --git a/src/backend/access/appendonly/appendonly_compaction.c b/src/backend/access/appendonly/appendonly_compaction.c
@@ -40,6 +40,7 @@
 #include "nodes/execnodes.h"
 #include "storage/procarray.h"
 #include "storage/lmgr.h"
+#include "utils/faultinjector.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/relcache.h"
@@ -518,7 +519,7 @@ AppendOnlyDrop(Relation aorel, List *compaction_segno)
 	int			i,
 				segno;
 	FileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
 
 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);
 	Assert(RelationIsAoRows(aorel));
@@ -571,7 +572,6 @@ AppendOnlyDrop(Relation aorel, List *compaction_segno)
 		FreeAllSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
 
 /*
@@ -589,7 +589,11 @@ AppendOnlyTruncateToEOF(Relation aorel)
 				segno;
 	LockAcquireResult acquireResult;
 	FileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	SIMPLE_FAULT_INJECTOR("ao_row_truncate_to_eof");
+#endif
 
 	Assert(RelationIsAoRows(aorel));
 
@@ -649,7 +653,6 @@ AppendOnlyTruncateToEOF(Relation aorel)
 		FreeAllSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
 
 /*
@@ -677,7 +680,11 @@ AppendOnlyCompact(Relation aorel,
 	int			i,
 				segno;
 	FileSegInfo *fsinfo;
-	Snapshot	appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	Snapshot	appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	SIMPLE_FAULT_INJECTOR("ao_row_compact");
+#endif
 
 	Assert(Gp_role == GP_ROLE_EXECUTE || Gp_role == GP_ROLE_UTILITY);
 	Assert(insert_segno >= 0);
@@ -753,7 +760,6 @@ AppendOnlyCompact(Relation aorel,
 		FreeAllSegFileInfo(segfile_array, total_segfiles);
 		pfree(segfile_array);
 	}
-	UnregisterSnapshot(appendOnlyMetaDataSnapshot);
 }
 
 /*

diff --git a/src/backend/access/appendonly/appendonlyam.c b/src/backend/access/appendonly/appendonlyam.c
@@ -2620,7 +2620,14 @@ appendonly_insert_init(Relation rel, int segno, bool update_mode)
 	 * Writers uses this since they have exclusive access to the lock acquired
 	 * with LockRelationAppendOnlySegmentFile for the segment-file.
 	 */
-	aoInsertDesc->appendOnlyMetaDataSnapshot = RegisterSnapshot(GetCatalogSnapshot(InvalidOid));
+	aoInsertDesc->appendOnlyMetaDataSnapshot = SnapshotSelf;
+
+#ifdef FAULT_INJECTOR
+	if (SIMPLE_FAULT_INJECTOR("ao_row_insert_init_1") == FaultInjectorTypeSkip)
+	{
+		SIMPLE_FAULT_INJECTOR("ao_row_insert_init_2");
+	}
+#endif
 
 	aoInsertDesc->mt_bind = create_memtuple_binding(RelationGetDescr(rel));
 
@@ -3093,8 +3100,6 @@ appendonly_insert_finish(AppendOnlyInsertDesc aoInsertDesc)
 
 	AppendOnlyStorageWrite_FinishSession(&aoInsertDesc->storageWrite);
 
-	UnregisterSnapshot(aoInsertDesc->appendOnlyMetaDataSnapshot);
-
 	pfree(aoInsertDesc->title);
 	pfree(aoInsertDesc);
 }

diff --git a/src/backend/access/appendonly/appendonlywriter.c b/src/backend/access/appendonly/appendonlywriter.c
@@ -1125,6 +1125,13 @@ SetSegnoForWrite(Relation rel, int existingsegno)
 	AORelHashEntryData *aoentry = NULL;
 	TransactionId CurrentXid = GetTopTransactionId();
 
+#ifdef FAULT_INJECTOR
+	if (SIMPLE_FAULT_INJECTOR("reserved_segno") == FaultInjectorTypeSkip)
+	{
+		return RESERVED_SEGNO;
+	}
+#endif
+
 	switch (Gp_role)
 	{
 		case GP_ROLE_EXECUTE:
@@ -1142,6 +1149,7 @@ SetSegnoForWrite(Relation rel, int existingsegno)
 
 			Assert(existingsegno == InvalidFileSegNumber);
 
+
 			ereportif(Debug_appendonly_print_segfile_choice, LOG,
 					  (errmsg("SetSegnoForWrite: Choosing a segno for append-only "
 							  "relation \"%s\" (%d) ",

diff --git a/src/backend/cdb/motion/ic_udpifc.c b/src/backend/cdb/motion/ic_udpifc.c
@@ -5888,16 +5888,12 @@ dispatcherAYT(void)
 static void
 checkQDConnectionAlive(void)
 {
-	if (!dispatcherAYT())
+	if (Gp_role == GP_ROLE_EXECUTE)
 	{
-		if (Gp_role == GP_ROLE_EXECUTE)
+		if (!dispatcherAYT())
 			ereport(ERROR,
 					(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
 					 errmsg("interconnect error segment lost contact with master (recv)")));
-		else
-			ereport(ERROR,
-					(errcode(ERRCODE_GP_INTERCONNECTION_ERROR),
-					 errmsg("interconnect error master lost contact with client (recv)")));
 	}
 }
 

diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c
@@ -69,6 +69,9 @@
 #include "postgres.h"
 #include <pthread.h>
 
+#ifdef HAVE_POLL_H
+#include <poll.h>
+#endif
 #include <signal.h>
 #include <fcntl.h>
 #include <grp.h>
@@ -2080,3 +2083,53 @@ pq_setkeepalivescount(int count, Port *port)
 
 	return STATUS_OK;
 }
+
+/*
+ * Check if the client is still connected.
+ */
+bool
+pq_check_connection(void)
+{
+	struct pollfd pollfd;
+	int         rc;
+	short		poll_ev_aux;
+
+#if defined(POLLRDHUP)
+	/*
+	 * POLLRDHUP is a Linux extension to poll(2) to detect sockets closed by the
+	 * other end.
+	 * We don't have a portable way to do that without actually trying to read
+	 * or write data on other systems. We don't want to read because that would
+	 * be confused by pipelined queries and COPY data. Perhaps in future we'll
+	 * try to write a heartbeat message instead.
+	 */
+	poll_ev_aux = POLLRDHUP;
+#elif defined(__darwin__)
+	/*
+	 * OSX is able to detect closed sockets via single POSIX-compliant POLLHUP
+	 * option
+	 */
+	poll_ev_aux = 0;
+#else
+	return true;
+#endif
+
+	pollfd.fd = MyProcPort->sock;
+	pollfd.events = POLLOUT | POLLIN | poll_ev_aux;
+
+	pollfd.revents = 0;
+
+	rc = poll(&pollfd, 1, 0);
+
+	if (rc < 0)
+	{
+		ereport(COMMERROR,
+				(errcode_for_socket_access(),
+				 errmsg("could not poll socket: %m")));
+		return false;
+	}
+	else if (rc == 1 && (pollfd.revents & (POLLHUP | poll_ev_aux)))
+		return false;
+
+	return true;
+}