patch proposal

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index d6c057a..f5a537c 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -252,6 +252,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -789,10 +790,12 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -4949,7 +4952,6 @@ readRecoveryCommandFile(void)
 			   *tail = NULL;
 	bool		recoveryTargetActionSet = false;
 
-
 	fd = AllocateFile(RECOVERY_COMMAND_FILE, "r");
 	if (fd == NULL)
 	{
@@ -5014,6 +5016,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5395,6 +5417,60 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the oldest xid of the
+	 * backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+								  ControlFile->checkPointCopy.oldestXid))
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_xid %u is older that backup's Xid %u", recoveryTargetXid, ControlFile->checkPointCopy.oldestXid),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/* Check if the recovery target lsn is prior to the latest checkpoint redo position of the backup */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_lsn \"%X/%X\" is older than backup start LSN \"%X/%X\"",
+							(uint32) (recoveryTargetLSN >> 32),
+							(uint32) recoveryTargetLSN, (uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/* Check if the recovery target time is prior to the current timestamp of the backup */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < ControlFile->checkPointCopy.time)
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_time %s is older than backup start time %s", timestamptz_to_str(recoveryTargetTime), timestamptz_to_str(ControlFile->checkPointCopy.time)),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -5740,6 +5816,23 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
+
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -6128,6 +6221,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -7040,6 +7136,46 @@ StartupXLOG(void)
 						break;
 				}
 			}
+			else
+			{
+				ereport(LOG,
+						(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+				errhint("This could be due to corrupt or missing WAL files.\n"
+						"All the WAL files needed for the recovery must be available to proceed to the recovery target "
+				"Or you might need to choose an earlier recovery target.")));
+
+				/*
+				 * This is the position where we can choose to shutdown, pause
+				 * or promote at the end-of-the-wal if the intended recovery
+				 * target is not reached
+				 */
+
+				switch (recoveryTargetIncomplete)
+				{
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+						/*
+						 * exit with special return code to request shutdown
+						 * of postmaster.  Log messages issued from
+						 * postmaster.
+						 */
+						ereport(LOG,
+								(errmsg("shutdown at end-of-the-wal")));
+
+						proc_exit(2);
+
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+						/* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+				}
+			}
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index ceb0462..799ddc2 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,18 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+}	RecoveryTargetIncomplete;
+
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 6cec027..ba417c4 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -252,6 +252,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -792,10 +793,12 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -5018,6 +5021,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5399,6 +5422,66 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the latest completed xid of the
+	 * backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+							 ControlFile->checkPointCopy.latestCompletedXid))
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_xid %u is older than the latest Completed Xid %u", recoveryTargetXid, ControlFile->checkPointCopy.latestCompletedXid),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target lsn is prior to the latest checkpointi's redo
+	 * position of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_lsn \"%X/%X\" is older than the backup start LSN \"%X/%X\"",
+							(uint32) (recoveryTargetLSN >> 32),
+							(uint32) recoveryTargetLSN, (uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target time is prior to the current timestamp of
+	 * the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < ControlFile->checkPointCopy.time)
+		{
+			ereport(FATAL,
+					(errmsg("recovery_target_time %s is older than the backup start time %s", timestamptz_to_str(recoveryTargetTime), timestamptz_to_str(ControlFile->checkPointCopy.time)),
+					 errhint("This means that the backup being used is much later than the recovery target position."
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -5744,6 +5827,23 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
+
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -6132,6 +6232,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -7044,6 +7147,44 @@ StartupXLOG(void)
 						break;
 				}
 			}
+		        else
+			{
+                              		ereport(LOG,
+							(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+							 errhint("This could be due to corrupt or missing WAL files.\n"
+								 "All the WAL files needed for the recovery must be available to proceed to the recovery target "
+								 "Or you might need to choose an earlier recovery target.")));
+
+                              /*
+                               * This is the position where we can choose to shutdown, pause
+                               * or promote at the end-of-the-wal if the intended recovery
+                               * target is not reached
+                               */
+                              switch (recoveryTargetIncomplete)
+                              {
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+					/*
+					 * exit with special return code to request shutdown
+					 * of postmaster.  Log messages issued from postmaster.
+                                         */
+					
+					ereport(LOG,
+							(errmsg("shutdown at end-of-the-wal")));
+					proc_exit(2);
+					
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+                                              /* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+                              }
+                      }
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
@@ -8450,6 +8591,7 @@ CreateCheckPoint(int flags)
 	checkPoint.nextXid = ShmemVariableCache->nextXid;
 	checkPoint.oldestXid = ShmemVariableCache->oldestXid;
 	checkPoint.oldestXidDB = ShmemVariableCache->oldestXidDB;
+	checkPoint.latestCompletedXid = ShmemVariableCache->latestCompletedXid;
 	LWLockRelease(XidGenLock);
 
 	LWLockAcquire(CommitTsLock, LW_SHARED);
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index 20077a6..d6d0852 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -192,114 +192,116 @@ main(int argc, char *argv[])
 	snprintf(sysident_str, sizeof(sysident_str), UINT64_FORMAT,
 			 ControlFile->system_identifier);
 
-	printf(_("pg_control version number:            %u\n"),
+	printf(_("pg_control version number:		%u\n"),
 		   ControlFile->pg_control_version);
-	printf(_("Catalog version number:               %u\n"),
+	printf(_("Catalog version number:                 %u\n"),
 		   ControlFile->catalog_version_no);
-	printf(_("Database system identifier:           %s\n"),
+	printf(_("Database system identifier:             %s\n"),
 		   sysident_str);
-	printf(_("Database cluster state:               %s\n"),
+	printf(_("Database cluster state:                 %s\n"),
 		   dbState(ControlFile->state));
-	printf(_("pg_control last modified:             %s\n"),
+	printf(_("pg_control last modified:               %s\n"),
 		   pgctime_str);
-	printf(_("Latest checkpoint location:           %X/%X\n"),
+	printf(_("Latest checkpoint location:             %X/%X\n"),
 		   (uint32) (ControlFile->checkPoint >> 32),
 		   (uint32) ControlFile->checkPoint);
-	printf(_("Prior checkpoint location:            %X/%X\n"),
+	printf(_("Prior checkpoint location:              %X/%X\n"),
 		   (uint32) (ControlFile->prevCheckPoint >> 32),
 		   (uint32) ControlFile->prevCheckPoint);
-	printf(_("Latest checkpoint's REDO location:    %X/%X\n"),
+	printf(_("Latest checkpoint's REDO location:      %X/%X\n"),
 		   (uint32) (ControlFile->checkPointCopy.redo >> 32),
 		   (uint32) ControlFile->checkPointCopy.redo);
-	printf(_("Latest checkpoint's REDO WAL file:    %s\n"),
+	printf(_("Latest checkpoint's REDO WAL file:      %s\n"),
 		   xlogfilename);
-	printf(_("Latest checkpoint's TimeLineID:       %u\n"),
+	printf(_("Latest checkpoint's TimeLineID:         %u\n"),
 		   ControlFile->checkPointCopy.ThisTimeLineID);
-	printf(_("Latest checkpoint's PrevTimeLineID:   %u\n"),
+	printf(_("Latest checkpoint's PrevTimeLineID:     %u\n"),
 		   ControlFile->checkPointCopy.PrevTimeLineID);
-	printf(_("Latest checkpoint's full_page_writes: %s\n"),
+	printf(_("Latest checkpoint's full_page_writes:   %s\n"),
 		   ControlFile->checkPointCopy.fullPageWrites ? _("on") : _("off"));
-	printf(_("Latest checkpoint's NextXID:          %u:%u\n"),
+	printf(_("Latest checkpoint's latestCompletedXID: %u\n"),
+		   ControlFile->checkPointCopy.latestCompletedXid);
+	printf(_("Latest checkpoint's NextXID:            %u:%u\n"),
 		   ControlFile->checkPointCopy.nextXidEpoch,
 		   ControlFile->checkPointCopy.nextXid);
-	printf(_("Latest checkpoint's NextOID:          %u\n"),
+	printf(_("Latest checkpoint's NextOID:            %u\n"),
 		   ControlFile->checkPointCopy.nextOid);
-	printf(_("Latest checkpoint's NextMultiXactId:  %u\n"),
+	printf(_("Latest checkpoint's NextMultiXactId:    %u\n"),
 		   ControlFile->checkPointCopy.nextMulti);
-	printf(_("Latest checkpoint's NextMultiOffset:  %u\n"),
+	printf(_("Latest checkpoint's NextMultiOffset:    %u\n"),
 		   ControlFile->checkPointCopy.nextMultiOffset);
-	printf(_("Latest checkpoint's oldestXID:        %u\n"),
+	printf(_("Latest checkpoint's oldestXID:          %u\n"),
 		   ControlFile->checkPointCopy.oldestXid);
-	printf(_("Latest checkpoint's oldestXID's DB:   %u\n"),
+	printf(_("Latest checkpoint's oldestXID's DB:     %u\n"),
 		   ControlFile->checkPointCopy.oldestXidDB);
-	printf(_("Latest checkpoint's oldestActiveXID:  %u\n"),
+	printf(_("Latest checkpoint's oldestActiveXID:    %u\n"),
 		   ControlFile->checkPointCopy.oldestActiveXid);
-	printf(_("Latest checkpoint's oldestMultiXid:   %u\n"),
+	printf(_("Latest checkpoint's oldestMultiXid:     %u\n"),
 		   ControlFile->checkPointCopy.oldestMulti);
-	printf(_("Latest checkpoint's oldestMulti's DB: %u\n"),
+	printf(_("Latest checkpoint's oldestMulti's DB:   %u\n"),
 		   ControlFile->checkPointCopy.oldestMultiDB);
-	printf(_("Latest checkpoint's oldestCommitTsXid:%u\n"),
+	printf(_("Latest checkpoint's oldestCommitTsXid:  %u\n"),
 		   ControlFile->checkPointCopy.oldestCommitTsXid);
-	printf(_("Latest checkpoint's newestCommitTsXid:%u\n"),
+	printf(_("Latest checkpoint's newestCommitTsXid:  %u\n"),
 		   ControlFile->checkPointCopy.newestCommitTsXid);
-	printf(_("Time of latest checkpoint:            %s\n"),
+	printf(_("Time of latest checkpoint:              %s\n"),
 		   ckpttime_str);
-	printf(_("Fake LSN counter for unlogged rels:   %X/%X\n"),
+	printf(_("Fake LSN counter for unlogged rels:     %X/%X\n"),
 		   (uint32) (ControlFile->unloggedLSN >> 32),
 		   (uint32) ControlFile->unloggedLSN);
-	printf(_("Minimum recovery ending location:     %X/%X\n"),
+	printf(_("Minimum recovery ending location:       %X/%X\n"),
 		   (uint32) (ControlFile->minRecoveryPoint >> 32),
 		   (uint32) ControlFile->minRecoveryPoint);
-	printf(_("Min recovery ending loc's timeline:   %u\n"),
+	printf(_("Min recovery ending loc's timeline:     %u\n"),
 		   ControlFile->minRecoveryPointTLI);
-	printf(_("Backup start location:                %X/%X\n"),
+	printf(_("Backup start location:                  %X/%X\n"),
 		   (uint32) (ControlFile->backupStartPoint >> 32),
 		   (uint32) ControlFile->backupStartPoint);
-	printf(_("Backup end location:                  %X/%X\n"),
+	printf(_("Backup end location:                    %X/%X\n"),
 		   (uint32) (ControlFile->backupEndPoint >> 32),
 		   (uint32) ControlFile->backupEndPoint);
-	printf(_("End-of-backup record required:        %s\n"),
+	printf(_("End-of-backup record required:          %s\n"),
 		   ControlFile->backupEndRequired ? _("yes") : _("no"));
-	printf(_("wal_level setting:                    %s\n"),
+	printf(_("wal_level setting:                      %s\n"),
 		   wal_level_str(ControlFile->wal_level));
-	printf(_("wal_log_hints setting:                %s\n"),
+	printf(_("wal_log_hints setting:                  %s\n"),
 		   ControlFile->wal_log_hints ? _("on") : _("off"));
-	printf(_("max_connections setting:              %d\n"),
+	printf(_("max_connections setting:                %d\n"),
 		   ControlFile->MaxConnections);
-	printf(_("max_worker_processes setting:         %d\n"),
+	printf(_("max_worker_processes setting:           %d\n"),
 		   ControlFile->max_worker_processes);
-	printf(_("max_prepared_xacts setting:           %d\n"),
+	printf(_("max_prepared_xacts setting:             %d\n"),
 		   ControlFile->max_prepared_xacts);
-	printf(_("max_locks_per_xact setting:           %d\n"),
+	printf(_("max_locks_per_xact setting:             %d\n"),
 		   ControlFile->max_locks_per_xact);
-	printf(_("track_commit_timestamp setting:       %s\n"),
+	printf(_("track_commit_timestamp setting:         %s\n"),
 		   ControlFile->track_commit_timestamp ? _("on") : _("off"));
-	printf(_("Maximum data alignment:               %u\n"),
+	printf(_("Maximum data alignment:                 %u\n"),
 		   ControlFile->maxAlign);
 	/* we don't print floatFormat since can't say much useful about it */
-	printf(_("Database block size:                  %u\n"),
+	printf(_("Database block size:                    %u\n"),
 		   ControlFile->blcksz);
-	printf(_("Blocks per segment of large relation: %u\n"),
+	printf(_("Blocks per segment of large relation:   %u\n"),
 		   ControlFile->relseg_size);
-	printf(_("WAL block size:                       %u\n"),
+	printf(_("WAL block size:                         %u\n"),
 		   ControlFile->xlog_blcksz);
-	printf(_("Bytes per WAL segment:                %u\n"),
+	printf(_("Bytes per WAL segment:                  %u\n"),
 		   ControlFile->xlog_seg_size);
-	printf(_("Maximum length of identifiers:        %u\n"),
+	printf(_("Maximum length of identifiers:          %u\n"),
 		   ControlFile->nameDataLen);
-	printf(_("Maximum columns in an index:          %u\n"),
+	printf(_("Maximum columns in an index:            %u\n"),
 		   ControlFile->indexMaxKeys);
-	printf(_("Maximum size of a TOAST chunk:        %u\n"),
+	printf(_("Maximum size of a TOAST chunk:          %u\n"),
 		   ControlFile->toast_max_chunk_size);
-	printf(_("Size of a large-object chunk:         %u\n"),
+	printf(_("Size of a large-object chunk:           %u\n"),
 		   ControlFile->loblksize);
-	printf(_("Date/time type storage:               %s\n"),
+	printf(_("Date/time type storage:                 %s\n"),
 		   (ControlFile->enableIntTimes ? _("64-bit integers") : _("floating-point numbers")));
-	printf(_("Float4 argument passing:              %s\n"),
+	printf(_("Float4 argument passing:                %s\n"),
 		   (ControlFile->float4ByVal ? _("by value") : _("by reference")));
-	printf(_("Float8 argument passing:              %s\n"),
+	printf(_("Float8 argument passing:                %s\n"),
 		   (ControlFile->float8ByVal ? _("by value") : _("by reference")));
-	printf(_("Data page checksum version:           %u\n"),
+	printf(_("Data page checksum version:             %u\n"),
 		   ControlFile->data_checksum_version);
 	return 0;
 }
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index ceb0462..35eabf2 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,18 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+} RecoveryTargetIncomplete;
+
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 0bc41ab..20da26b 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -21,7 +21,7 @@
 
 
 /* Version identifier for this pg_control format */
-#define PG_CONTROL_VERSION	960
+#define PG_CONTROL_VERSION	100
 
 /*
  * Body of CheckPoint XLOG records.  This is declared here because we keep
@@ -58,6 +58,7 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+	TransactionId latestCompletedXid;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index 8c24ae2..59ef116 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -211,6 +211,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the checkpoint time of the backup being used is earlier than the specified recovery_target_time. 
+        In other words, the recovery will not proceed further if the checkpoint time of the backup is later than the specified recovery_target_time.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -231,6 +236,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the latest completed xid of the backup is earlier than the specified recovery target xid.
+        In other words, the recovery will not proceed further if the latest completed xid of the backup is later than the specified target xid.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -248,6 +258,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         parameter is parsed using the system data type
         <link linkend="datatype-pg-lsn"><type>pg_lsn</></link>.
        </para>
+       <para>
+        This parameter also checks if the current checkpoint's redo position of the backup is earlier than the specified recovery target lsn.
+        In other words, the recovery will not proceed further if the latest checkpoint's redo position of the backup is later than the specified target lsn.
+       </para>
+
       </listitem>
      </varlistentry>
      </variablelist>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8379133..3043864 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -826,6 +826,7 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
@@ -5449,6 +5450,67 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the latest completed xid
+	 * of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+							 ControlFile->checkPointCopy.latestCompletedXid))
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_xid %u is older than the latest Completed Xid %u", recoveryTargetXid, ControlFile->checkPointCopy.latestCompletedXid),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target lsn is prior to the latest checkpointi's
+	 * redo position of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_lsn \"%X/%X\" is older than the backup start LSN \"%X/%X\"",
+							(uint32) (recoveryTargetLSN >> 32),
+							(uint32) recoveryTargetLSN, (uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target time is prior to the current timestamp of
+	 * the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < ControlFile->checkPointCopy.time)
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_time %s is older than the backup start time %s", timestamptz_to_str(recoveryTargetTime),
+					   timestamptz_to_str(ControlFile->checkPointCopy.time)),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -6182,6 +6244,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -8531,6 +8596,7 @@ CreateCheckPoint(int flags)
 	checkPoint.nextXid = ShmemVariableCache->nextXid;
 	checkPoint.oldestXid = ShmemVariableCache->oldestXid;
 	checkPoint.oldestXidDB = ShmemVariableCache->oldestXidDB;
+	checkPoint.latestCompletedXid = ShmemVariableCache->latestCompletedXid;
 	LWLockRelease(XidGenLock);
 
 	LWLockAcquire(CommitTsLock, LW_SHARED);
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index 20077a6..f340f1c 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -222,6 +222,8 @@ main(int argc, char *argv[])
 	printf(_("Latest checkpoint's NextXID:          %u:%u\n"),
 		   ControlFile->checkPointCopy.nextXidEpoch,
 		   ControlFile->checkPointCopy.nextXid);
+	printf(_("Latest checkpoint's latestCompletedXID: %u\n"),
+		   ControlFile->checkPointCopy.latestCompletedXid);
 	printf(_("Latest checkpoint's NextOID:          %u\n"),
 		   ControlFile->checkPointCopy.nextOid);
 	printf(_("Latest checkpoint's NextMultiXactId:  %u\n"),
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 23731e9..7ed9484 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -21,7 +21,7 @@
 
 
 /* Version identifier for this pg_control format */
-#define PG_CONTROL_VERSION	960
+#define PG_CONTROL_VERSION	100
 
 /*
  * Body of CheckPoint XLOG records.  This is declared here because we keep
@@ -58,6 +58,7 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+	TransactionId latestCompletedXid;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index 8c24ae2..fa580b7 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -349,6 +349,48 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>
 
+     <varlistentry id="recovery-target-incomplete"
+                   xreflabel="recovery_target_incomplete">
+      <term><varname>recovery_target_incomplete</varname> (<type>enum</type>)
+      <indexterm>
+        <primary><varname>recovery_target_incomplete</> recovery parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies what action the server should take once the recovery target is
+        not reached. The default is <literal>promote</>, which means recovery will
+        finish and the server will start to accept connections. <literal>pause</> means the recovery process will be paused.
+        Finally <literal>shutdown</> will stop the server if the recovery cannot proceed any further.
+       </para>
+       <para>
+        The intended use of the <literal>pause</> setting is to allow queries
+        to be executed against the database to check if this recovery target
+        is the most desirable point for recovery.
+        The paused state can be resumed by
+        using <function>pg_xlog_replay_resume()</> (see
+        <xref linkend="functions-recovery-control-table">), which then
+        causes recovery to end. If this recovery target is not the
+        desired stopping point, then shut down the server, change the
+        recovery target settings to a later target and restart to
+        continue recovery.
+       </para>
+       <para>
+        The <literal>shutdown</> setting is useful to have the instance ready
+        at the exact replay point desired.  The instance will still be able to
+        replay more WAL records (and in fact will have to replay WAL records
+        since the last checkpoint next time it is started).
+       </para>
+       <para>
+        Note that because <filename>recovery.conf</> will not be renamed when
+        <varname>recovery_target_incomplete</> is set to <literal>shutdown</>,
+        any subsequent start will end with immediate shutdown unless the
+        configuration is changed or the <filename>recovery.conf</> file is
+        removed manually.
+       </para>
+      </listitem>
+     </varlistentry>
+
     </variablelist>
    </sect1>
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8379133..343f5b6 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -252,6 +252,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -830,6 +831,7 @@ static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -5068,6 +5070,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5794,6 +5816,21 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -7094,6 +7131,46 @@ StartupXLOG(void)
 						break;
 				}
 			}
+			else
+			{
+				ereport(LOG,
+						(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+				errhint("This could be due to corrupt or missing WAL files.\n"
+						"All the WAL files needed for the recovery must be available to proceed to the recovery target "
+				"Or you might need to choose an earlier recovery target.")));
+
+				/*
+				 * This is the position where we can choose to shutdown, pause
+				 * or promote at the end-of-the-wal if the intended recovery
+				 * target is not reached
+				 */
+				switch (recoveryTargetIncomplete)
+				{
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+						/*
+						 * exit with special return code to request shutdown
+						 * of postmaster.  Log messages issued from
+						 * postmaster.
+						 */
+
+						ereport(LOG,
+								(errmsg("shutdown at end-of-the-wal")));
+						proc_exit(2);
+
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+						/* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+				}
+
+			}
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index 8ad4d47..e46f50d 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,17 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+}	RecoveryTargetIncomplete;
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..73ef387 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -211,6 +211,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the checkpoint time of the backup being used is earlier than the specified recovery_target_time. 
+        In other words, the recovery will not proceed further if the checkpoint time of the backup is later than the specified recovery_target_time.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -231,6 +236,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the latest completed xid of the backup is earlier than the specified recovery target xid.
+        In other words, the recovery will not proceed further if the latest completed xid of the backup is later than the specified target xid.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -248,6 +258,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         parameter is parsed using the system data type
         <link linkend="datatype-pg-lsn"><type>pg_lsn</></link>.
        </para>
+       <para>
+        This parameter also checks if the current checkpoint's redo position of the backup is earlier than the specified recovery target lsn.
+        In other words, the recovery will not proceed further if the latest checkpoint's redo position of the backup is later than the specified target lsn.
+       </para>
+
       </listitem>
      </varlistentry>
      </variablelist>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 5016273..e8411a0 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -834,6 +834,7 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
@@ -5544,6 +5545,67 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the latest completed xid
+	 * of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+							 ControlFile->checkPointCopy.latestCompletedXid))
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_xid %u is older than the latest Completed Xid %u", recoveryTargetXid, ControlFile->checkPointCopy.latestCompletedXid),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target lsn is prior to the latest checkpointi's
+	 * redo position of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_lsn \"%X/%X\" is older than the backup start LSN \"%X/%X\"",
+							(uint32) (recoveryTargetLSN >> 32),
+							(uint32) recoveryTargetLSN, (uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target time is prior to the current timestamp of
+	 * the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < ControlFile->checkPointCopy.time)
+		{
+			ereport(ERROR,
+					(errmsg("recovery_target_time %s is older than the backup start time %s", timestamptz_to_str(recoveryTargetTime),
+					   timestamptz_to_str(ControlFile->checkPointCopy.time)),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -6277,6 +6339,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -8642,6 +8707,7 @@ CreateCheckPoint(int flags)
 	checkPoint.nextXid = ShmemVariableCache->nextXid;
 	checkPoint.oldestXid = ShmemVariableCache->oldestXid;
 	checkPoint.oldestXidDB = ShmemVariableCache->oldestXidDB;
+	checkPoint.latestCompletedXid = ShmemVariableCache->latestCompletedXid;
 	LWLockRelease(XidGenLock);
 
 	LWLockAcquire(CommitTsLock, LW_SHARED);
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index f47171d..5ba170c 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -222,6 +222,8 @@ main(int argc, char *argv[])
 	printf(_("Latest checkpoint's NextXID:          %u:%u\n"),
 		   ControlFile->checkPointCopy.nextXidEpoch,
 		   ControlFile->checkPointCopy.nextXid);
+	printf(_("Latest checkpoint's latestCompletedXID: %u\n"),
+		   ControlFile->checkPointCopy.latestCompletedXid);
 	printf(_("Latest checkpoint's NextOID:          %u\n"),
 		   ControlFile->checkPointCopy.nextOid);
 	printf(_("Latest checkpoint's NextMultiXactId:  %u\n"),
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index e4194b9..a8c9101 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -58,6 +58,7 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+	TransactionId latestCompletedXid;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 4018f0a..fb0a48b 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -601,6 +601,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -618,6 +623,7 @@ sub init_from_backup
 	$params{hba_permit_replication} = 1
 	  unless defined $params{hba_permit_replication};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -641,6 +647,7 @@ port = $port
 	$self->set_replication_conf         if $params{hba_permit_replication};
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -673,6 +680,31 @@ sub start
 	$self->_update_pid;
 }
 
+
+=pod
+
+=item $node->start_pitr()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub start_pitr
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+        my $log    = $self->logfile;
+        print("### Starting node \"$name\"\n");
+        my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+	return $ret;
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -789,7 +821,6 @@ sub enable_restoring
 	  $TestLib::windows_os
 	  ? qq{copy "$path\\\\%f" "%p"}
 	  : qq{cp "$path/%f" "%p"};
-
 	$self->append_conf(
 		'recovery.conf', qq(
 restore_command = '$copy_command'
@@ -797,6 +828,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..9aa7a6c 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 11;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -45,6 +45,64 @@ sub test_recovery_standby
 	$node_standby->teardown_node;
 }
 
+# Test recovery start point for various recovery targets
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+	my $num_rows	    = shift;
+
+        my $node_standby = get_new_node($node_name);
+        $node_standby->init_from_backup($node_master, 'my_pitr',
+                has_restoring_pitr => 1);
+
+        foreach my $param_item (@$recovery_params)
+        {
+                $node_standby->append_conf(
+                        'recovery.conf',
+                                qq($param_item
+));
+        }
+        my $ret = $node_standby->start_pitr;
+
+	my $log = $node_standby->logfile;
+	my $irx = "is older than the latest Completed Xid";
+        my $irl = "is older than the backup start LSN";
+
+        my @rx = `grep -inr "$irx" $log`;
+        my @rl = `grep -inr "$irl" $log`;
+
+        print "matching lines for xid: @rx\n";
+        print "matching lines for lsn : @rl\n";
+
+        if ($ret != 0)
+        {
+                if (@rx)
+                {
+                        print "Recovery could not start as the specified target XID was not legitimate. Test Passed\n";
+                        isnt($ret,0, "Successful $test_name.");
+                        goto TEST_RECOVERY_LSN;
+
+                }
+                elsif(@rl)
+                {
+			print "Recovery could not start as the specified target LSN was not legitimate. Test Passed\n";
+			isnt($ret,0, "Successful $test_name.");
+			exit 0;
+
+                }
+                else
+                {
+                        print "# pg_ctl failed; logfile:\n";
+                        print TestLib::slurp_file($node_standby->logfile);
+                        BAIL_OUT("pg_ctl failed");
+                }
+        }
+}
+
 # Initialize master node
 my $node_master = get_new_node('master');
 $node_master->init(has_archiving => 1, allows_streaming => 1);
@@ -144,3 +202,51 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Check if the recovery start point is legitimate
+# If the recovery start point is not appropriate, recovery should not start at all
+
+# Generate some data on master to advance the database position
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(4001,5000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(5001,6000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(6001,7000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(7001,8000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(8001,9000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(9001,10000))");
+
+print "required data generated.\n";
+
+my $rxid = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn10, $recovery_lxid) = split /\|/, $rxid;
+
+# Now take a fresh backup
+
+$node_master->backup('my_pitr');
+
+print "Old LSN prior to my_pitr is : $lsn1\n";
+print "Old XID prior to my_pitr is : $recovery_txid\n";
+print "Old TIMESTAMP prior to the pitr is : $recovery_time\n";
+
+# Test the recovery start point for XID. Specify an old XID which falls prior to the backup's latest XID
+@recovery_params = (
+         "recovery_target_xid  = '$recovery_txid'",);
+
+test_recovery_pitr('Recovery Start Point check for XID',
+                 'pitr_1', $node_master, \@recovery_params, "10000");
+
+# Test the checking of the recovery start point for target LSN. Specify an old LSN which falls prior to the backup's latest LSN.
+TEST_RECOVERY_LSN:
+my $recovery_olsn = $lsn1;
+@recovery_params = (
+         "recovery_target_lsn  = '$recovery_olsn'",);
+
+test_recovery_pitr('Recovery Start Point check for LSN',
+                 'pitr_2', $node_master, \@recovery_params, "20000");

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..9053bb0 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -349,6 +349,48 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>
 
+     <varlistentry id="recovery-target-incomplete"
+                   xreflabel="recovery_target_incomplete">
+      <term><varname>recovery_target_incomplete</varname> (<type>enum</type>)
+      <indexterm>
+        <primary><varname>recovery_target_incomplete</> recovery parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies what action the server should take once the recovery target is
+        not reached. The default is <literal>promote</>, which means recovery will
+        finish and the server will start to accept connections. <literal>pause</> means the recovery process will be paused.
+        Finally <literal>shutdown</> will stop the server if the recovery cannot proceed any further.
+       </para>
+       <para>
+        The intended use of the <literal>pause</> setting is to allow queries
+        to be executed against the database to check if this recovery target
+        is the most desirable point for recovery.
+        The paused state can be resumed by
+        using <function>pg_xlog_replay_resume()</> (see
+        <xref linkend="functions-recovery-control-table">), which then
+        causes recovery to end. If this recovery target is not the
+        desired stopping point, then shut down the server, change the
+        recovery target settings to a later target and restart to
+        continue recovery.
+       </para>
+       <para>
+        The <literal>shutdown</> setting is useful to have the instance ready
+        at the exact replay point desired.  The instance will still be able to
+        replay more WAL records (and in fact will have to replay WAL records
+        since the last checkpoint next time it is started).
+       </para>
+       <para>
+        Note that because <filename>recovery.conf</> will not be renamed when
+        <varname>recovery_target_incomplete</> is set to <literal>shutdown</>,
+        any subsequent start will end with immediate shutdown unless the
+        configuration is changed or the <filename>recovery.conf</> file is
+        removed manually.
+       </para>
+      </listitem>
+     </varlistentry>
+
     </variablelist>
    </sect1>
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 5016273..159fe16 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -259,6 +259,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -838,6 +839,7 @@ static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -5163,6 +5165,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5889,6 +5911,21 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -7205,6 +7242,46 @@ StartupXLOG(void)
 						break;
 				}
 			}
+			else
+			{
+				ereport(LOG,
+						(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+				errhint("This could be due to corrupt or missing WAL files.\n"
+						"All the WAL files needed for the recovery must be available to proceed to the recovery target "
+				"Or you might need to choose an earlier recovery target.")));
+
+				/*
+				 * This is the position where we can choose to shutdown, pause
+				 * or promote at the end-of-the-wal if the intended recovery
+				 * target is not reached
+				 */
+				switch (recoveryTargetIncomplete)
+				{
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+						/*
+						 * exit with special return code to request shutdown
+						 * of postmaster.  Log messages issued from
+						 * postmaster.
+						 */
+
+						ereport(LOG,
+								(errmsg("shutdown at end-of-the-wal")));
+						proc_exit(2);
+
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+						/* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+				}
+
+			}
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index 578bff5..b9d707c 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,17 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+}	RecoveryTargetIncomplete;
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 4018f0a..e8f6969 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -601,6 +601,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -618,6 +623,7 @@ sub init_from_backup
 	$params{hba_permit_replication} = 1
 	  unless defined $params{hba_permit_replication};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -641,6 +647,7 @@ port = $port
 	$self->set_replication_conf         if $params{hba_permit_replication};
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -673,6 +680,68 @@ sub start
 	$self->_update_pid;
 }
 
+=item $node->pitr_start()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub pitr_start
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+	my $log	   = $self->logfile;
+	print("### Starting node \"$name\"\n");
+	my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+        if ($ret != 0)
+        {
+                my $s = "shutdown at end-of-the-wal";
+
+                my @rs = `grep $s, $log`;
+
+                if (@rs)
+                {
+                        print "Database was shutdown at end-of-the-wal. Test Passed\n";
+			isnt($ret,0, "check the PITR node for successful SHUTDOWN at incomplete recovery");
+			exit 0;
+
+                }
+		else
+		{
+			print "# pg_ctl failed; logfile:\n";
+			print TestLib::slurp_file($self->logfile);
+			BAIL_OUT("pg_ctl failed");
+		}
+        }
+
+        $self->_update_pid;
+}
+
+=item $node->status()
+
+Wrapper for pg_ctl status
+
+Check the node status
+
+=cut
+
+sub status
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+        print("### Checking the node status \"$name\"\n");
+        my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+                $self->logfile, 'status');
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -789,7 +858,6 @@ sub enable_restoring
 	  $TestLib::windows_os
 	  ? qq{copy "$path\\\\%f" "%p"}
 	  : qq{cp "$path/%f" "%p"};
-
 	$self->append_conf(
 		'recovery.conf', qq(
 restore_command = '$copy_command'
@@ -797,6 +865,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..e47fb74 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 12;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -27,9 +27,7 @@ sub test_recovery_standby
 			qq($param_item
 ));
 	}
-
 	$node_standby->start;
-
 	# Wait until standby has replayed enough data
 	my $caughtup_query =
 	  "SELECT '$until_lsn'::pg_lsn <= pg_last_wal_replay_location()";
@@ -41,6 +39,44 @@ sub test_recovery_standby
 	  $node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
 	is($result, qq($num_rows), "check standby content for $test_name");
 
+        print "Rows in PITR  : $result\n";
+        print "Rows in master: $num_rows\n";
+
+	# Stop standby node
+	$node_standby->teardown_node;
+}
+
+
+# Create a node from given backup to perform PITR to a certain recovery target.
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+        my $num_rows        = shift;
+        my $until_lsn       = shift;
+
+	my $node_standby = get_new_node($node_name);
+	$node_standby->init_from_backup($node_master, 'my_pitr',
+		has_restoring_pitr => 1);
+
+	foreach my $param_item (@$recovery_params)
+	{
+		$node_standby->append_conf(
+			'recovery.conf',
+				qq($param_item
+));
+	}
+	$node_standby->pitr_start;
+	# Check the content on pitr
+	my $result =
+		$node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
+	isnt($result, qq($num_rows), "check the PITR node for successful $test_name");
+	print "Rows in PITR  : $result\n";
+	print "Rows in master: $num_rows\n";
+
 	# Stop standby node
 	$node_standby->teardown_node;
 }
@@ -144,3 +180,79 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Test Incomplete recovery
+# Initialize master node
+
+$node_master = get_new_node('master1');
+$node_master->init(has_archiving => 1, allows_streaming => 1);
+
+# Start it
+$node_master->start;
+
+# Create data before taking the backup, aimed at testing
+
+$node_master->safe_psql('postgres',
+         "CREATE TABLE tab_int AS SELECT generate_series(1,1000) AS a");
+
+	my $initial_lsn = $node_master->safe_psql('postgres', "SELECT pg_current_wal_location();");
+	print "LSN Before backup: $initial_lsn\n";
+	my $initial_xid = $node_master->safe_psql('postgres', "SELECT txid_current();");
+	print "XID before backup: $initial_xid\n";
+	my $initial_rows = $node_master->safe_psql('postgres', "SELECT count(*) from tab_int;");
+	print "Initial rows before backup: $initial_rows\n";
+
+# Take backup from which all operations will be run
+$node_master->backup('my_pitr');
+
+# Generate enough data and more WAL Archives for a recovery target reference.
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(1001,10000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(10001,20000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(20001,30000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(30001,40000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(40001,60000))");
+
+# current wal position in the master node
+$ret = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn6, $recovery_xid) = split /\|/, $ret;
+
+# Test the server promotion when the recovery fails to reach the recovery target.
+my $recovery_target_incomplete='promote';
+
+@recovery_params = (
+	"recovery_target_xid = '$recovery_xid'",
+	"recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PROMOTION at incomplete recovery','pitr_1',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server pauses when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='pause';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PAUSE at incomplete recovery','pitr_2',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server successfully shuts down when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='shutdown';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('SHUTDOWN at incomplete recovery','pitr_3',$node_master, \@recovery_params, "60000", $lsn6);
diff --git a/src/tools/pgindent/pgindent b/src/tools/pgindent/pgindent
index 0d3859d..f90a449 100755
--- a/src/tools/pgindent/pgindent
+++ b/src/tools/pgindent/pgindent
@@ -247,7 +247,7 @@ sub pre_indent
 		# previous line was struct, etc.
 		next
 		  if $srclines[ $lno - 1 ] =~
-			  m!=|^(struct|enum|[ \t]*typedef|extern[ \t]+"C")!;
+		  m!=|^(struct|enum|[ \t]*typedef|extern[ \t]+"C")!;
 
 		$srclines[$lno] = "$l2\nint pgindent_func_no_var_fix;";
 	}
@@ -520,7 +520,7 @@ File::Find::find(
 			  && -f _
 			  && /^.*\.[ch]\z/s
 			  && push(@files, $File::Find::name);
-		  }
+		}
 	},
 	$code_base) if $code_base;
 
diff --git a/src/tools/pgindent/pgperltidy b/src/tools/pgindent/pgperltidy
index 6098e18..395bce3 100755
--- a/src/tools/pgindent/pgperltidy
+++ b/src/tools/pgindent/pgperltidy
@@ -17,7 +17,7 @@ PERLTIDY=${PERLTIDY:-perltidy}
 	cut -d: -f1
 ) |
 sort -u |
-xargs $PERLTIDY --profile=src/tools/pgindent/perltidyrc
+xargs $PERLTIDY --profile=/data/PostgreSQL-Repo/postgresql/src/tools/pgindent/perltidyrc
 
 # perltidyrc specifies --backup-and-modify-in-place, so get rid of .bak files
 find . -type f -name '*.bak' | xargs rm

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..73ef387 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -211,6 +211,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the checkpoint time of the backup being used is earlier than the specified recovery_target_time. 
+        In other words, the recovery will not proceed further if the checkpoint time of the backup is later than the specified recovery_target_time.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -231,6 +236,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the latest completed xid of the backup is earlier than the specified recovery target xid.
+        In other words, the recovery will not proceed further if the latest completed xid of the backup is later than the specified target xid.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -248,6 +258,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         parameter is parsed using the system data type
         <link linkend="datatype-pg-lsn"><type>pg_lsn</></link>.
        </para>
+       <para>
+        This parameter also checks if the current checkpoint's redo position of the backup is earlier than the specified recovery target lsn.
+        In other words, the recovery will not proceed further if the latest checkpoint's redo position of the backup is later than the specified target lsn.
+       </para>
+
       </listitem>
      </varlistentry>
      </variablelist>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8973583..f030baa 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -834,6 +834,7 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
@@ -5544,6 +5545,66 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the latest completed xid
+	 * of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+							 ControlFile->checkPointCopy.latestCompletedXid))
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_xid is older than the latest Completed Xid which is %u", ControlFile->checkPointCopy.latestCompletedXid),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target lsn is prior to the latest checkpointi's
+	 * redo position of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_lsn is older than the backup start LSN which is \"%X/%X\"",
+								(uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target time is prior to the current timestamp of
+	 * the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < time_t_to_timestamptz(ControlFile->checkPointCopy.time))
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_time is older than the backup's latest checkpoint time which is %s",
+											pstrdup(str_time(ControlFile->checkPointCopy.time))),
+					errhint("This means that the backup being used is much later than the recovery target position.\n"
+						"You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -6277,6 +6338,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -8642,6 +8706,7 @@ CreateCheckPoint(int flags)
 	checkPoint.nextXid = ShmemVariableCache->nextXid;
 	checkPoint.oldestXid = ShmemVariableCache->oldestXid;
 	checkPoint.oldestXidDB = ShmemVariableCache->oldestXidDB;
+	checkPoint.latestCompletedXid = ShmemVariableCache->latestCompletedXid;
 	LWLockRelease(XidGenLock);
 
 	LWLockAcquire(CommitTsLock, LW_SHARED);
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index f47171d..5ba170c 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -222,6 +222,8 @@ main(int argc, char *argv[])
 	printf(_("Latest checkpoint's NextXID:          %u:%u\n"),
 		   ControlFile->checkPointCopy.nextXidEpoch,
 		   ControlFile->checkPointCopy.nextXid);
+	printf(_("Latest checkpoint's latestCompletedXID: %u\n"),
+		   ControlFile->checkPointCopy.latestCompletedXid);
 	printf(_("Latest checkpoint's NextOID:          %u\n"),
 		   ControlFile->checkPointCopy.nextOid);
 	printf(_("Latest checkpoint's NextMultiXactId:  %u\n"),
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index e4194b9..a8c9101 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -58,6 +58,7 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+	TransactionId latestCompletedXid;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 4018f0a..92c5367 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -601,6 +601,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -618,6 +623,7 @@ sub init_from_backup
 	$params{hba_permit_replication} = 1
 	  unless defined $params{hba_permit_replication};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -641,6 +647,7 @@ port = $port
 	$self->set_replication_conf         if $params{hba_permit_replication};
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -673,6 +680,31 @@ sub start
 	$self->_update_pid;
 }
 
+
+=pod
+
+=item $node->start_pitr()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub start_pitr
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+        my $log    = $self->logfile;
+        print("### Starting node \"$name\"\n");
+        my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+	return $ret;
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -797,6 +829,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..ec6b9fe 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 12;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -45,6 +45,69 @@ sub test_recovery_standby
 	$node_standby->teardown_node;
 }
 
+# Test recovery start point for various recovery targets
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+	my $num_rows	    = shift;
+
+        my $node_standby = get_new_node($node_name);
+        $node_standby->init_from_backup($node_master, 'my_pitr',
+                has_restoring_pitr => 1);
+
+        foreach my $param_item (@$recovery_params)
+        {
+                $node_standby->append_conf(
+                        'recovery.conf',
+                                qq($param_item
+));
+        }
+        my $ret = $node_standby->start_pitr;
+
+	my $log = $node_standby->logfile;
+	my $irx = "Specified recovery_target_xid is older than the latest Completed Xid";
+        my $irl = "Specified recovery_target_lsn is older than the backup start LSN";
+	my $irt = "Specified recovery_target_time is older than the backup's latest checkpoint time";
+
+        my @rx = `grep -inr "$irx" $log`;
+        my @rl = `grep -inr "$irl" $log`;
+	my @rt = `grep -inr "$irt" $log`;
+
+        if ($ret != 0)
+        {
+                if (@rx)
+                {
+                        print "Recovery could not start as the specified target XID was not legitimate. Test Passed\n";
+                        isnt($ret,0, "Successful $test_name.");
+                        goto TEST_RECOVERY_LSN;
+
+                }
+                elsif(@rl)
+                {
+			print "Recovery could not start as the specified target LSN was not legitimate. Test Passed\n";
+			isnt($ret,0, "Successful $test_name.");
+			goto TEST_RECOVERY_TIME;
+
+                }
+		elsif(@rt)
+		{
+                        print "Recovery could not start as the specified target TIME was not legitimate. Test Passed\n";
+                        isnt($ret,0, "Successful $test_name.");
+                        exit 0;
+		}
+                else
+                {
+                        print "# pg_ctl failed; logfile:\n";
+                        print TestLib::slurp_file($node_standby->logfile);
+                        BAIL_OUT("pg_ctl failed");
+                }
+        }
+}
+
 # Initialize master node
 my $node_master = get_new_node('master');
 $node_master->init(has_archiving => 1, allows_streaming => 1);
@@ -144,3 +207,57 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Check if the recovery start point is legitimate
+# If the recovery start point is not appropriate, recovery should not start at all
+
+# Generate some data on master to advance the database position
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(4001,5000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(5001,6000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(6001,7000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(7001,8000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(8001,9000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(9001,10000))");
+
+print "required data generated.\n";
+
+my $rxid = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn10, $recovery_lxid) = split /\|/, $rxid;
+
+# Now take a fresh backup
+
+$node_master->backup('my_pitr');
+
+# Test the recovery start point for XID. Specify an old XID which falls prior to the backup's latest XID
+@recovery_params = (
+         "recovery_target_xid  = '$recovery_txid'",);
+
+test_recovery_pitr('Recovery Start Point check for XID',
+                 'pitr_1', $node_master, \@recovery_params, "10000");
+
+# Test the checking of the recovery start point for target LSN. Specify an old recovery target LSN which falls prior to the backup's latest LSN.
+TEST_RECOVERY_LSN:
+my $recovery_olsn = $lsn1;
+@recovery_params = (
+         "recovery_target_lsn  = '$recovery_olsn'",);
+
+test_recovery_pitr('Recovery Start Point check for LSN',
+                 'pitr_2', $node_master, \@recovery_params, "10000");
+
+# Test the checking of the recovery start point for target TIME. Specify an old recovery target TIME which falls prior to the backup's latest Timestamp.
+TEST_RECOVERY_TIME:
+
+my $recovery_otime = $recovery_time;
+@recovery_params = (
+         "recovery_target_time  = '$recovery_otime'",);
+
+test_recovery_pitr('Recovery Start Point check for TIME',
+                 'pitr_3', $node_master, \@recovery_params, "10000");

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..9053bb0 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -349,6 +349,48 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>
 
+     <varlistentry id="recovery-target-incomplete"
+                   xreflabel="recovery_target_incomplete">
+      <term><varname>recovery_target_incomplete</varname> (<type>enum</type>)
+      <indexterm>
+        <primary><varname>recovery_target_incomplete</> recovery parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies what action the server should take once the recovery target is
+        not reached. The default is <literal>promote</>, which means recovery will
+        finish and the server will start to accept connections. <literal>pause</> means the recovery process will be paused.
+        Finally <literal>shutdown</> will stop the server if the recovery cannot proceed any further.
+       </para>
+       <para>
+        The intended use of the <literal>pause</> setting is to allow queries
+        to be executed against the database to check if this recovery target
+        is the most desirable point for recovery.
+        The paused state can be resumed by
+        using <function>pg_xlog_replay_resume()</> (see
+        <xref linkend="functions-recovery-control-table">), which then
+        causes recovery to end. If this recovery target is not the
+        desired stopping point, then shut down the server, change the
+        recovery target settings to a later target and restart to
+        continue recovery.
+       </para>
+       <para>
+        The <literal>shutdown</> setting is useful to have the instance ready
+        at the exact replay point desired.  The instance will still be able to
+        replay more WAL records (and in fact will have to replay WAL records
+        since the last checkpoint next time it is started).
+       </para>
+       <para>
+        Note that because <filename>recovery.conf</> will not be renamed when
+        <varname>recovery_target_incomplete</> is set to <literal>shutdown</>,
+        any subsequent start will end with immediate shutdown unless the
+        configuration is changed or the <filename>recovery.conf</> file is
+        removed manually.
+       </para>
+      </listitem>
+     </varlistentry>
+
     </variablelist>
    </sect1>
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8973583..7257a49 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -259,6 +259,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -838,6 +839,7 @@ static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -5163,6 +5165,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5889,6 +5911,21 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -7205,6 +7242,46 @@ StartupXLOG(void)
 						break;
 				}
 			}
+			else
+			{
+				ereport(LOG,
+						(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+				errhint("This could be due to corrupt or missing WAL files.\n"
+						"All the WAL files needed for the recovery must be available to proceed to the recovery target "
+				"Or you might need to choose an earlier recovery target.")));
+
+				/*
+				 * This is the position where we can choose to shutdown, pause
+				 * or promote at the end-of-the-wal if the intended recovery
+				 * target is not reached
+				 */
+				switch (recoveryTargetIncomplete)
+				{
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+						/*
+						 * exit with special return code to request shutdown
+						 * of postmaster.  Log messages issued from
+						 * postmaster.
+						 */
+
+						ereport(LOG,
+								(errmsg("shutdown at end-of-the-wal")));
+						proc_exit(2);
+
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+						/* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+				}
+
+			}
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index 578bff5..b9d707c 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,17 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+}	RecoveryTargetIncomplete;
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 4018f0a..119d43d 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -601,6 +601,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -618,6 +623,7 @@ sub init_from_backup
 	$params{hba_permit_replication} = 1
 	  unless defined $params{hba_permit_replication};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -641,6 +647,7 @@ port = $port
 	$self->set_replication_conf         if $params{hba_permit_replication};
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -673,6 +680,49 @@ sub start
 	$self->_update_pid;
 }
 
+=item $node->start_pitr()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub start_pitr
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+	my $log	   = $self->logfile;
+	print("### Starting node \"$name\"\n");
+	my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+        if ($ret != 0)
+        {
+                my $s = "shutdown at end-of-the-wal";
+
+                my @rs = `grep $s, $log`;
+
+                if (@rs)
+                {
+                        print "Database was shutdown at end-of-the-wal. Test Passed\n";
+			isnt($ret,0, "check the PITR node for successful SHUTDOWN at incomplete recovery");
+			exit 0;
+
+                }
+		else
+		{
+			print "# pg_ctl failed; logfile:\n";
+			print TestLib::slurp_file($self->logfile);
+			BAIL_OUT("pg_ctl failed");
+		}
+        }
+
+        $self->_update_pid;
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -797,6 +847,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..e8e9d14 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 12;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -27,9 +27,7 @@ sub test_recovery_standby
 			qq($param_item
 ));
 	}
-
 	$node_standby->start;
-
 	# Wait until standby has replayed enough data
 	my $caughtup_query =
 	  "SELECT '$until_lsn'::pg_lsn <= pg_last_wal_replay_location()";
@@ -41,6 +39,44 @@ sub test_recovery_standby
 	  $node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
 	is($result, qq($num_rows), "check standby content for $test_name");
 
+        print "Rows in PITR  : $result\n";
+        print "Rows in master: $num_rows\n";
+
+	# Stop standby node
+	$node_standby->teardown_node;
+}
+
+
+# Create a node from given backup to perform PITR to a certain recovery target.
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+        my $num_rows        = shift;
+        my $until_lsn       = shift;
+
+	my $node_standby = get_new_node($node_name);
+	$node_standby->init_from_backup($node_master, 'my_pitr',
+		has_restoring_pitr => 1);
+
+	foreach my $param_item (@$recovery_params)
+	{
+		$node_standby->append_conf(
+			'recovery.conf',
+				qq($param_item
+));
+	}
+	$node_standby->start_pitr;
+	# Check the content on pitr
+	my $result =
+		$node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
+	isnt($result, qq($num_rows), "check the PITR node for successful $test_name");
+	print "Rows in PITR  : $result\n";
+	print "Rows in master: $num_rows\n";
+
 	# Stop standby node
 	$node_standby->teardown_node;
 }
@@ -144,3 +180,72 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Test Incomplete recovery
+# Initialize master node
+
+$node_master = get_new_node('master1');
+$node_master->init(has_archiving => 1, allows_streaming => 1);
+
+# Start it
+$node_master->start;
+
+# Create data before taking the backup, aimed at testing
+
+$node_master->safe_psql('postgres',
+         "CREATE TABLE tab_int AS SELECT generate_series(1,1000) AS a");
+
+# Take backup from which all operations will be run
+$node_master->backup('my_pitr');
+
+# Generate enough data and more WAL Archives for a recovery target reference.
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(1001,10000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(10001,20000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(20001,30000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(30001,40000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(40001,60000))");
+
+# current wal position in the master node
+$ret = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn6, $recovery_xid) = split /\|/, $ret;
+
+# Test the server promotion when the recovery fails to reach the recovery target.
+my $recovery_target_incomplete='promote';
+
+@recovery_params = (
+	"recovery_target_xid = '$recovery_xid'",
+	"recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PROMOTION at incomplete recovery','pitr_1',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server pauses when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='pause';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PAUSE at incomplete recovery','pitr_2',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server successfully shuts down when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='shutdown';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('SHUTDOWN at incomplete recovery','pitr_3',$node_master, \@recovery_params, "60000", $lsn6);

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..73ef387 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -211,6 +211,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the checkpoint time of the backup being used is earlier than the specified recovery_target_time. 
+        In other words, the recovery will not proceed further if the checkpoint time of the backup is later than the specified recovery_target_time.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -231,6 +236,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         The precise stopping point is also influenced by
         <xref linkend="recovery-target-inclusive">.
        </para>
+       <para>
+        This parameter also checks if the latest completed xid of the backup is earlier than the specified recovery target xid.
+        In other words, the recovery will not proceed further if the latest completed xid of the backup is later than the specified target xid.
+       </para>
+
       </listitem>
      </varlistentry>
 
@@ -248,6 +258,11 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
         parameter is parsed using the system data type
         <link linkend="datatype-pg-lsn"><type>pg_lsn</></link>.
        </para>
+       <para>
+        This parameter also checks if the current checkpoint's redo position of the backup is earlier than the specified recovery target lsn.
+        In other words, the recovery will not proceed further if the latest checkpoint's redo position of the backup is later than the specified target lsn.
+       </para>
+
       </listitem>
      </varlistentry>
      </variablelist>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 58790e0..787a61a 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -841,6 +841,7 @@ static MemoryContext walDebugCxt = NULL;
 #endif
 
 static void readRecoveryCommandFile(void);
+static void recoveryStartsHere(void);
 static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
@@ -5601,6 +5602,66 @@ getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
 	return false;
 }
 
+/* When performing point-in-time-recovery, this function identifies if
+ * the specified recovery target (recovery_target_time, recovery_target_lsn and recovery_target_xid) is prior to that of the backup.
+ * Which means, recovery cannot proceed if the recovery target point is prior to backup start point.
+ */
+
+static void
+recoveryStartsHere(void)
+{
+	/*
+	 * Check if the recovery target xid is older than the latest completed xid
+	 * of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_XID)
+	{
+		if (TransactionIdPrecedes(recoveryTargetXid,
+							 ControlFile->checkPointCopy.latestCompletedXid))
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_xid is older than the latest Completed Xid which is %u", ControlFile->checkPointCopy.latestCompletedXid),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target lsn is prior to the latest checkpointi's
+	 * redo position of the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_LSN)
+	{
+		if (recoveryTargetLSN < ControlFile->checkPointCopy.redo)
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_lsn is older than the backup start LSN which is \"%X/%X\"",
+								(uint32) (ControlFile->checkPointCopy.redo >> 32), (uint32) ControlFile->checkPointCopy.redo),
+					 errhint("This means that the backup being used is much later than the recovery target position.\n"
+							 "You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+
+	/*
+	 * Check if the recovery target time is prior to the current timestamp of
+	 * the backup
+	 */
+
+	if (recoveryTarget == RECOVERY_TARGET_TIME)
+	{
+		if (recoveryTargetTime < time_t_to_timestamptz(ControlFile->checkPointCopy.time))
+		{
+			ereport(ERROR,
+					(errmsg("Specified recovery_target_time is older than the backup's latest checkpoint time which is %s",
+											pstrdup(str_time(ControlFile->checkPointCopy.time))),
+					errhint("This means that the backup being used is much later than the recovery target position.\n"
+						"You might need to use a backup taken prior to the recovery target point.")));
+		}
+	}
+}
+
 /*
  * For point-in-time recovery, this function decides whether we want to
  * stop applying the XLOG before the current record.
@@ -6334,6 +6395,9 @@ StartupXLOG(void)
 					(errmsg("starting archive recovery")));
 	}
 
+	/* Check if archive recovery can start at all */
+	recoveryStartsHere();
+
 	/*
 	 * Take ownership of the wakeup latch if we're going to sleep during
 	 * recovery.
@@ -8705,6 +8769,7 @@ CreateCheckPoint(int flags)
 	checkPoint.nextXid = ShmemVariableCache->nextXid;
 	checkPoint.oldestXid = ShmemVariableCache->oldestXid;
 	checkPoint.oldestXidDB = ShmemVariableCache->oldestXidDB;
+	checkPoint.latestCompletedXid = ShmemVariableCache->latestCompletedXid;
 	LWLockRelease(XidGenLock);
 
 	LWLockAcquire(CommitTsLock, LW_SHARED);
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index 2ea8931..d70a221 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -228,6 +228,8 @@ main(int argc, char *argv[])
 	printf(_("Latest checkpoint's NextXID:          %u:%u\n"),
 		   ControlFile->checkPointCopy.nextXidEpoch,
 		   ControlFile->checkPointCopy.nextXid);
+	printf(_("Latest checkpoint's latestCompletedXID: %u\n"),
+		   ControlFile->checkPointCopy.latestCompletedXid);
 	printf(_("Latest checkpoint's NextOID:          %u\n"),
 		   ControlFile->checkPointCopy.nextOid);
 	printf(_("Latest checkpoint's NextMultiXactId:  %u\n"),
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 3a25cc8..898cec7 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -60,6 +60,7 @@ typedef struct CheckPoint
 	 * set to InvalidTransactionId.
 	 */
 	TransactionId oldestActiveXid;
+	TransactionId latestCompletedXid;
 } CheckPoint;
 
 /* XLOG info values for XLOG rmgr */
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 5ef007f..a448443 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -589,6 +589,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -604,6 +609,7 @@ sub init_from_backup
 
 	$params{has_streaming} = 0 unless defined $params{has_streaming};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -626,6 +632,7 @@ port = $port
 ));
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -658,6 +665,31 @@ sub start
 	$self->_update_pid;
 }
 
+
+=pod
+
+=item $node->start_pitr()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub start_pitr
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+        my $log    = $self->logfile;
+        print("### Starting node \"$name\"\n");
+        my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+	return $ret;
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -782,6 +814,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..ec6b9fe 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 12;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -45,6 +45,69 @@ sub test_recovery_standby
 	$node_standby->teardown_node;
 }
 
+# Test recovery start point for various recovery targets
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+	my $num_rows	    = shift;
+
+        my $node_standby = get_new_node($node_name);
+        $node_standby->init_from_backup($node_master, 'my_pitr',
+                has_restoring_pitr => 1);
+
+        foreach my $param_item (@$recovery_params)
+        {
+                $node_standby->append_conf(
+                        'recovery.conf',
+                                qq($param_item
+));
+        }
+        my $ret = $node_standby->start_pitr;
+
+	my $log = $node_standby->logfile;
+	my $irx = "Specified recovery_target_xid is older than the latest Completed Xid";
+        my $irl = "Specified recovery_target_lsn is older than the backup start LSN";
+	my $irt = "Specified recovery_target_time is older than the backup's latest checkpoint time";
+
+        my @rx = `grep -inr "$irx" $log`;
+        my @rl = `grep -inr "$irl" $log`;
+	my @rt = `grep -inr "$irt" $log`;
+
+        if ($ret != 0)
+        {
+                if (@rx)
+                {
+                        print "Recovery could not start as the specified target XID was not legitimate. Test Passed\n";
+                        isnt($ret,0, "Successful $test_name.");
+                        goto TEST_RECOVERY_LSN;
+
+                }
+                elsif(@rl)
+                {
+			print "Recovery could not start as the specified target LSN was not legitimate. Test Passed\n";
+			isnt($ret,0, "Successful $test_name.");
+			goto TEST_RECOVERY_TIME;
+
+                }
+		elsif(@rt)
+		{
+                        print "Recovery could not start as the specified target TIME was not legitimate. Test Passed\n";
+                        isnt($ret,0, "Successful $test_name.");
+                        exit 0;
+		}
+                else
+                {
+                        print "# pg_ctl failed; logfile:\n";
+                        print TestLib::slurp_file($node_standby->logfile);
+                        BAIL_OUT("pg_ctl failed");
+                }
+        }
+}
+
 # Initialize master node
 my $node_master = get_new_node('master');
 $node_master->init(has_archiving => 1, allows_streaming => 1);
@@ -144,3 +207,57 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Check if the recovery start point is legitimate
+# If the recovery start point is not appropriate, recovery should not start at all
+
+# Generate some data on master to advance the database position
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(4001,5000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(5001,6000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(6001,7000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(7001,8000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(8001,9000))");
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(9001,10000))");
+
+print "required data generated.\n";
+
+my $rxid = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn10, $recovery_lxid) = split /\|/, $rxid;
+
+# Now take a fresh backup
+
+$node_master->backup('my_pitr');
+
+# Test the recovery start point for XID. Specify an old XID which falls prior to the backup's latest XID
+@recovery_params = (
+         "recovery_target_xid  = '$recovery_txid'",);
+
+test_recovery_pitr('Recovery Start Point check for XID',
+                 'pitr_1', $node_master, \@recovery_params, "10000");
+
+# Test the checking of the recovery start point for target LSN. Specify an old recovery target LSN which falls prior to the backup's latest LSN.
+TEST_RECOVERY_LSN:
+my $recovery_olsn = $lsn1;
+@recovery_params = (
+         "recovery_target_lsn  = '$recovery_olsn'",);
+
+test_recovery_pitr('Recovery Start Point check for LSN',
+                 'pitr_2', $node_master, \@recovery_params, "10000");
+
+# Test the checking of the recovery start point for target TIME. Specify an old recovery target TIME which falls prior to the backup's latest Timestamp.
+TEST_RECOVERY_TIME:
+
+my $recovery_otime = $recovery_time;
+@recovery_params = (
+         "recovery_target_time  = '$recovery_otime'",);
+
+test_recovery_pitr('Recovery Start Point check for TIME',
+                 'pitr_3', $node_master, \@recovery_params, "10000");

diff --git a/doc/src/sgml/recovery-config.sgml b/doc/src/sgml/recovery-config.sgml
index a91864b..9053bb0 100644
--- a/doc/src/sgml/recovery-config.sgml
+++ b/doc/src/sgml/recovery-config.sgml
@@ -349,6 +349,48 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>
 
+     <varlistentry id="recovery-target-incomplete"
+                   xreflabel="recovery_target_incomplete">
+      <term><varname>recovery_target_incomplete</varname> (<type>enum</type>)
+      <indexterm>
+        <primary><varname>recovery_target_incomplete</> recovery parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Specifies what action the server should take once the recovery target is
+        not reached. The default is <literal>promote</>, which means recovery will
+        finish and the server will start to accept connections. <literal>pause</> means the recovery process will be paused.
+        Finally <literal>shutdown</> will stop the server if the recovery cannot proceed any further.
+       </para>
+       <para>
+        The intended use of the <literal>pause</> setting is to allow queries
+        to be executed against the database to check if this recovery target
+        is the most desirable point for recovery.
+        The paused state can be resumed by
+        using <function>pg_xlog_replay_resume()</> (see
+        <xref linkend="functions-recovery-control-table">), which then
+        causes recovery to end. If this recovery target is not the
+        desired stopping point, then shut down the server, change the
+        recovery target settings to a later target and restart to
+        continue recovery.
+       </para>
+       <para>
+        The <literal>shutdown</> setting is useful to have the instance ready
+        at the exact replay point desired.  The instance will still be able to
+        replay more WAL records (and in fact will have to replay WAL records
+        since the last checkpoint next time it is started).
+       </para>
+       <para>
+        Note that because <filename>recovery.conf</> will not be renamed when
+        <varname>recovery_target_incomplete</> is set to <literal>shutdown</>,
+        any subsequent start will end with immediate shutdown unless the
+        configuration is changed or the <filename>recovery.conf</> file is
+        removed manually.
+       </para>
+      </listitem>
+     </varlistentry>
+
     </variablelist>
    </sect1>
 
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 58790e0..4969519 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -260,6 +260,7 @@ static char *archiveCleanupCommand = NULL;
 static RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;
 static bool recoveryTargetInclusive = true;
 static RecoveryTargetAction recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;
+static RecoveryTargetIncomplete recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
 static TransactionId recoveryTargetXid;
 static TimestampTz recoveryTargetTime;
 static char *recoveryTargetName;
@@ -845,6 +846,7 @@ static void exitArchiveRecovery(TimeLineID endTLI, XLogRecPtr endOfLog);
 static bool recoveryStopsBefore(XLogReaderState *record);
 static bool recoveryStopsAfter(XLogReaderState *record);
 static void recoveryPausesHere(void);
+static void IncompleteRecoveryPause(void);
 static bool recoveryApplyDelay(XLogReaderState *record);
 static void SetLatestXTime(TimestampTz xtime);
 static void SetCurrentChunkStartTime(TimestampTz xtime);
@@ -5220,6 +5222,26 @@ readRecoveryCommandFile(void)
 
 			recoveryTargetActionSet = true;
 		}
+		else if (strcmp(item->name, "recovery_target_incomplete") == 0)
+		{
+			if (strcmp(item->value, "pause") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PAUSE;
+			else if (strcmp(item->value, "promote") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_PROMOTE;
+			else if (strcmp(item->value, "shutdown") == 0)
+				recoveryTargetIncomplete = RECOVERY_TARGET_INCOMPLETE_SHUTDOWN;
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				errmsg("invalid value for recovery parameter \"%s\": \"%s\"",
+					   "recovery_target_incomplete",
+					   item->value),
+						 errhint("Valid values are \"pause\", \"promote\", and \"shutdown\".")));
+
+			ereport(DEBUG2,
+					(errmsg_internal("recovery_target_incomplete = '%s'",
+									 item->value)));
+		}
 		else if (strcmp(item->name, "recovery_target_timeline") == 0)
 		{
 			rtliGiven = true;
@@ -5946,6 +5968,21 @@ SetRecoveryPause(bool recoveryPause)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+static void
+IncompleteRecoveryPause(void)
+{
+	/* Pause recovery at end-of-the-wal when recovery target is not reached */
+	ereport(LOG,
+			(errmsg("recovery has paused"),
+			 errhint("Execute pg_xlog_replay_resume() to continue.")));
+
+	while (RecoveryIsPaused())
+	{
+		pg_usleep(1000000L);	/* 1000 ms */
+		HandleStartupProcInterrupts();
+	}
+}
+
 /*
  * When recovery_min_apply_delay is set, we wait long enough to make sure
  * certain record types are applied at least that interval behind the master.
@@ -7263,6 +7300,46 @@ StartupXLOG(void)
 						break;
 				}
 			}
+			else
+			{
+				ereport(LOG,
+						(errmsg("recovery has reached end-of-the-wal and has not reached the recovery target yet"),
+				errhint("This could be due to corrupt or missing WAL files.\n"
+						"All the WAL files needed for the recovery must be available to proceed to the recovery target "
+				"Or you might need to choose an earlier recovery target.")));
+
+				/*
+				 * This is the position where we can choose to shutdown, pause
+				 * or promote at the end-of-the-wal if the intended recovery
+				 * target is not reached
+				 */
+				switch (recoveryTargetIncomplete)
+				{
+
+					case RECOVERY_TARGET_INCOMPLETE_SHUTDOWN:
+
+						/*
+						 * exit with special return code to request shutdown
+						 * of postmaster.  Log messages issued from
+						 * postmaster.
+						 */
+
+						ereport(LOG,
+								(errmsg("shutdown at end-of-the-wal")));
+						proc_exit(2);
+
+					case RECOVERY_TARGET_INCOMPLETE_PAUSE:
+
+						SetRecoveryPause(true);
+						IncompleteRecoveryPause();
+
+						/* drop into promote */
+
+					case RECOVERY_TARGET_INCOMPLETE_PROMOTE:
+						break;
+				}
+
+			}
 
 			/* Allow resource managers to do any required cleanup. */
 			for (rmid = 0; rmid <= RM_MAX_ID; rmid++)
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h
index c09c0f8..380c3ee 100644
--- a/src/include/access/xlog_internal.h
+++ b/src/include/access/xlog_internal.h
@@ -256,6 +256,17 @@ typedef enum
 } RecoveryTargetAction;
 
 /*
+ * Recovery target incomplete.
+ */
+
+typedef enum
+{
+	RECOVERY_TARGET_INCOMPLETE_PAUSE,
+	RECOVERY_TARGET_INCOMPLETE_PROMOTE,
+	RECOVERY_TARGET_INCOMPLETE_SHUTDOWN
+}	RecoveryTargetIncomplete;
+
+/*
  * Method table for resource managers.
  *
  * This struct must be kept in sync with the PG_RMGR definition in
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index 5ef007f..4e18e78 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -589,6 +589,11 @@ Restoring WAL segments from archives using restore_command can be enabled
 by passing the keyword parameter has_restoring => 1. This is disabled by
 default.
 
+Restoring WAL segments from archives using restore command to perform PITR
+can be enabled by passing the keyword parameter has_restoring_pitr => 1. This
+is disabled by default. By enabling this parameter, the standby database will
+not operate in standby mode.
+
 The backup is copied, leaving the original unmodified. pg_hba.conf is
 unconditionally set to enable replication connections.
 
@@ -604,6 +609,7 @@ sub init_from_backup
 
 	$params{has_streaming} = 0 unless defined $params{has_streaming};
 	$params{has_restoring} = 0 unless defined $params{has_restoring};
+	$params{has_restoring_pitr} = 0 unless defined $params{has_restoring_pitr};
 
 	print
 "# Initializing node \"$node_name\" from backup \"$backup_name\" of node \"$root_name\"\n";
@@ -626,6 +632,7 @@ port = $port
 ));
 	$self->enable_streaming($root_node) if $params{has_streaming};
 	$self->enable_restoring($root_node) if $params{has_restoring};
+	$self->enable_restoring_pitr($root_node) if $params{has_restoring_pitr};
 }
 
 =pod
@@ -658,6 +665,68 @@ sub start
 	$self->_update_pid;
 }
 
+=item $node->pitr_start()
+
+Wrapper for pg_ctl start
+
+Start the node and wait until it is ready to accept connections.
+
+=cut
+
+sub pitr_start
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+	my $log	   = $self->logfile;
+	print("### Starting node \"$name\"\n");
+	my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+               $self->logfile, 'start');
+
+        if ($ret != 0)
+        {
+                my $s = "shutdown at end-of-the-wal";
+
+                my @rs = `grep $s, $log`;
+
+                if (@rs)
+                {
+                        print "Database was shutdown at end-of-the-wal. Test Passed\n";
+			isnt($ret,0, "check the PITR node for successful SHUTDOWN at incomplete recovery");
+			exit 0;
+
+                }
+		else
+		{
+			print "# pg_ctl failed; logfile:\n";
+			print TestLib::slurp_file($self->logfile);
+			BAIL_OUT("pg_ctl failed");
+		}
+        }
+
+        $self->_update_pid;
+}
+
+=item $node->status()
+
+Wrapper for pg_ctl status
+
+Check the node status
+
+=cut
+
+sub status
+{
+        my ($self) = @_;
+        my $port   = $self->port;
+        my $pgdata = $self->data_dir;
+        my $name   = $self->name;
+        print("### Checking the node status \"$name\"\n");
+        my $ret = TestLib::system_log('pg_ctl', '-D', $self->data_dir, '-l',
+                $self->logfile, 'status');
+}
+
 =pod
 
 =item $node->stop(mode)
@@ -774,7 +843,6 @@ sub enable_restoring
 	  $TestLib::windows_os
 	  ? qq{copy "$path\\\\%f" "%p"}
 	  : qq{cp "$path/%f" "%p"};
-
 	$self->append_conf(
 		'recovery.conf', qq(
 restore_command = '$copy_command'
@@ -782,6 +850,32 @@ standby_mode = on
 ));
 }
 
+# Internal routine to enable archive recovery command on a standby node for PITR
+sub enable_restoring_pitr
+{
+        my ($self, $root_node) = @_;
+        my $path = $root_node->archive_dir;
+        my $name = $self->name;
+
+        print "### Enabling WAL restore for node \"$name\"\n";
+
+        # On Windows, the path specified in the restore command needs to use
+        # double back-slashes to work properly and to be able to detect properly
+        # the file targeted by the copy command, so the directory value used
+        # in this routine, using only one back-slash, need to be properly changed
+        # first. Paths also need to be double-quoted to prevent failures where
+        # the path contains spaces.
+        $path =~ s{\\}{\\\\}g if ($TestLib::windows_os);
+        my $copy_command =
+          $TestLib::windows_os
+          ? qq{copy "$path\\\\%f" "%p"}
+          : qq{cp "$path/%f" "%p"};
+	$self->append_conf(
+		'recovery.conf', qq(
+restore_command = '$copy_command'
+));
+}
+
 # Internal routine to enable archiving
 sub enable_archiving
 {
diff --git a/src/test/recovery/t/003_recovery_targets.pl b/src/test/recovery/t/003_recovery_targets.pl
index b7b0caa..e47fb74 100644
--- a/src/test/recovery/t/003_recovery_targets.pl
+++ b/src/test/recovery/t/003_recovery_targets.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 9;
+use Test::More tests => 12;
 
 # Create and test a standby from given backup, with a certain
 # recovery target.
@@ -27,9 +27,7 @@ sub test_recovery_standby
 			qq($param_item
 ));
 	}
-
 	$node_standby->start;
-
 	# Wait until standby has replayed enough data
 	my $caughtup_query =
 	  "SELECT '$until_lsn'::pg_lsn <= pg_last_wal_replay_location()";
@@ -41,6 +39,44 @@ sub test_recovery_standby
 	  $node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
 	is($result, qq($num_rows), "check standby content for $test_name");
 
+        print "Rows in PITR  : $result\n";
+        print "Rows in master: $num_rows\n";
+
+	# Stop standby node
+	$node_standby->teardown_node;
+}
+
+
+# Create a node from given backup to perform PITR to a certain recovery target.
+
+sub test_recovery_pitr
+{
+        my $test_name       = shift;
+        my $node_name       = shift;
+        my $node_master     = shift;
+        my $recovery_params = shift;
+        my $num_rows        = shift;
+        my $until_lsn       = shift;
+
+	my $node_standby = get_new_node($node_name);
+	$node_standby->init_from_backup($node_master, 'my_pitr',
+		has_restoring_pitr => 1);
+
+	foreach my $param_item (@$recovery_params)
+	{
+		$node_standby->append_conf(
+			'recovery.conf',
+				qq($param_item
+));
+	}
+	$node_standby->pitr_start;
+	# Check the content on pitr
+	my $result =
+		$node_standby->safe_psql('postgres', "SELECT count(*) FROM tab_int");
+	isnt($result, qq($num_rows), "check the PITR node for successful $test_name");
+	print "Rows in PITR  : $result\n";
+	print "Rows in master: $num_rows\n";
+
 	# Stop standby node
 	$node_standby->teardown_node;
 }
@@ -144,3 +180,79 @@ test_recovery_standby('XID + time + name',
 	"recovery_target_lsn = '$recovery_lsn'",);
 test_recovery_standby('XID + time + name + LSN',
 	'standby_9', $node_master, \@recovery_params, "5000", $lsn5);
+
+# Test Incomplete recovery
+# Initialize master node
+
+$node_master = get_new_node('master1');
+$node_master->init(has_archiving => 1, allows_streaming => 1);
+
+# Start it
+$node_master->start;
+
+# Create data before taking the backup, aimed at testing
+
+$node_master->safe_psql('postgres',
+         "CREATE TABLE tab_int AS SELECT generate_series(1,1000) AS a");
+
+	my $initial_lsn = $node_master->safe_psql('postgres', "SELECT pg_current_wal_location();");
+	print "LSN Before backup: $initial_lsn\n";
+	my $initial_xid = $node_master->safe_psql('postgres', "SELECT txid_current();");
+	print "XID before backup: $initial_xid\n";
+	my $initial_rows = $node_master->safe_psql('postgres', "SELECT count(*) from tab_int;");
+	print "Initial rows before backup: $initial_rows\n";
+
+# Take backup from which all operations will be run
+$node_master->backup('my_pitr');
+
+# Generate enough data and more WAL Archives for a recovery target reference.
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(1001,10000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(10001,20000))");
+# Force archiving of WAL file
+$node_master->safe_psql('postgres', "SELECT pg_switch_wal()");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(20001,30000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(30001,40000))");
+
+$node_master->safe_psql('postgres',
+        "INSERT INTO tab_int VALUES (generate_series(40001,60000))");
+
+# current wal position in the master node
+$ret = $node_master->safe_psql('postgres',
+        "SELECT pg_current_wal_location(), txid_current();");
+my ($lsn6, $recovery_xid) = split /\|/, $ret;
+
+# Test the server promotion when the recovery fails to reach the recovery target.
+my $recovery_target_incomplete='promote';
+
+@recovery_params = (
+	"recovery_target_xid = '$recovery_xid'",
+	"recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PROMOTION at incomplete recovery','pitr_1',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server pauses when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='pause';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('PAUSE at incomplete recovery','pitr_2',$node_master, \@recovery_params, "60000", $lsn6);
+
+# Check if the server successfully shuts down when the recovery stops mid-way without reaching the recovery target.
+
+$recovery_target_incomplete='shutdown';
+
+@recovery_params = (
+        "recovery_target_xid = '$recovery_xid'",
+        "recovery_target_incomplete = '$recovery_target_incomplete'");
+test_recovery_pitr('SHUTDOWN at incomplete recovery','pitr_3',$node_master, \@recovery_params, "60000", $lsn6);
diff --git a/src/tools/pgindent/pgindent b/src/tools/pgindent/pgindent
index 0d3859d..f90a449 100755
--- a/src/tools/pgindent/pgindent
+++ b/src/tools/pgindent/pgindent
@@ -247,7 +247,7 @@ sub pre_indent
 		# previous line was struct, etc.
 		next
 		  if $srclines[ $lno - 1 ] =~
-			  m!=|^(struct|enum|[ \t]*typedef|extern[ \t]+"C")!;
+		  m!=|^(struct|enum|[ \t]*typedef|extern[ \t]+"C")!;
 
 		$srclines[$lno] = "$l2\nint pgindent_func_no_var_fix;";
 	}
@@ -520,7 +520,7 @@ File::Find::find(
 			  && -f _
 			  && /^.*\.[ch]\z/s
 			  && push(@files, $File::Find::name);
-		  }
+		}
 	},
 	$code_base) if $code_base;
 
diff --git a/src/tools/pgindent/pgperltidy b/src/tools/pgindent/pgperltidy
index 6098e18..395bce3 100755
--- a/src/tools/pgindent/pgperltidy
+++ b/src/tools/pgindent/pgperltidy
@@ -17,7 +17,7 @@ PERLTIDY=${PERLTIDY:-perltidy}
 	cut -d: -f1
 ) |
 sort -u |
-xargs $PERLTIDY --profile=src/tools/pgindent/perltidyrc
+xargs $PERLTIDY --profile=/data/PostgreSQL-Repo/postgresql/src/tools/pgindent/perltidyrc
 
 # perltidyrc specifies --backup-and-modify-in-place, so get rid of .bak files
 find . -type f -name '*.bak' | xargs rm