Implement waiting for given lsn at transaction start

This commit adds following optional clause to BEGIN and START TRANSACTION
commands.

  WAIT FOR LSN lsn [ TIMEOUT timeout ]

New clause pospones transaction start till given lsn is applied on standby.
This clause allows user be sure, that changes previously made on primary would
be visible on standby.

New shared memory struct is used to track awaited lsn per backend.  Recovery
process wakes up backend once required lsn is applied.

Author: Ivan Kartyshov, Anna Akenteva
Reviewed-by: Craig Ringer, Thomas Munro, Robert Haas, Kyotaro Horiguchi
Reviewed-by: Masahiko Sawada, Ants Aasma, Dmitry Ivanov, Simon Riggs
Reviewed-by: Amit Kapila, Alexander Korotkov
Discussion: https://postgr.es/m/0240c26c-9f84-30ea-fca9-93ab2df5f305%40postgrespro.ru

3 files changed, 56 insertions, 0 deletions

diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 1525c0de725..db179becab5 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -3748,10 +3748,22 @@ _copyTransactionStmt(const TransactionStmt *from)
 	COPY_STRING_FIELD(savepoint_name);
 	COPY_STRING_FIELD(gid);
 	COPY_SCALAR_FIELD(chain);
+	COPY_NODE_FIELD(wait);
 
 	return newnode;
 }
 
+static WaitClause *
+_copyWaitClause(const WaitClause *from)
+{
+	WaitClause *newnode = makeNode(WaitClause);
+
+	COPY_STRING_FIELD(lsn);
+	COPY_SCALAR_FIELD(timeout);
+
+	return newnode;
+};
+
 static CompositeTypeStmt *
 _copyCompositeTypeStmt(const CompositeTypeStmt *from)
 {
@@ -5339,6 +5351,9 @@ copyObjectImpl(const void *from)
 		case T_TransactionStmt:
 			retval = _copyTransactionStmt(from);
 			break;
+		case T_WaitClause:
+			retval = _copyWaitClause(from);
+			break;
 		case T_CompositeTypeStmt:
 			retval = _copyCompositeTypeStmt(from);
 			break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 4f34189ab5c..854d484f603 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1541,6 +1541,16 @@ _equalTransactionStmt(const TransactionStmt *a, const TransactionStmt *b)
 	COMPARE_STRING_FIELD(savepoint_name);
 	COMPARE_STRING_FIELD(gid);
 	COMPARE_SCALAR_FIELD(chain);
+	COMPARE_NODE_FIELD(wait);
+
+	return true;
+}
+
+static bool
+_equalWaitClause(const WaitClause *a, const WaitClause *b)
+{
+	COMPARE_STRING_FIELD(lsn);
+	COMPARE_SCALAR_FIELD(timeout);
 
 	return true;
 }
@@ -3391,6 +3401,9 @@ equal(const void *a, const void *b)
 		case T_TransactionStmt:
 			retval = _equalTransactionStmt(a, b);
 			break;
+		case T_WaitClause:
+			retval = _equalWaitClause(a, b);
+			break;
 		case T_CompositeTypeStmt:
 			retval = _equalCompositeTypeStmt(a, b);
 			break;
diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c
index 5b826509ebe..47753b42c69 100644
--- a/src/backend/nodes/outfuncs.c
+++ b/src/backend/nodes/outfuncs.c
@@ -2785,6 +2785,28 @@ _outDefElem(StringInfo str, const DefElem *node)
 }
 
 static void
+_outTransactionStmt(StringInfo str, const TransactionStmt *node)
+{
+	WRITE_NODE_TYPE("TRANSACTIONSTMT");
+
+	WRITE_STRING_FIELD(savepoint_name);
+	WRITE_STRING_FIELD(gid);
+	WRITE_NODE_FIELD(options);
+	WRITE_BOOL_FIELD(chain);
+	WRITE_ENUM_FIELD(kind, TransactionStmtKind);
+	WRITE_NODE_FIELD(wait);
+}
+
+static void
+_outWaitClause(StringInfo str, const WaitClause *node)
+{
+	WRITE_NODE_TYPE("WAITCLAUSE");
+
+	WRITE_STRING_FIELD(lsn);
+	WRITE_UINT_FIELD(timeout);
+}
+
+static void
 _outTableLikeClause(StringInfo str, const TableLikeClause *node)
 {
 	WRITE_NODE_TYPE("TABLELIKECLAUSE");
@@ -4334,6 +4356,12 @@ outNode(StringInfo str, const void *obj)
 			case T_PartitionRangeDatum:
 				_outPartitionRangeDatum(str, obj);
 				break;
+			case T_TransactionStmt:
+				_outTransactionStmt(str, obj);
+				break;
+			case T_WaitClause:
+				_outWaitClause(str, obj);
+				break;
 
 			default: