Make validate logic like restore. issue #5

Author: stalkerg

parsexlog.c

@@ -30,6 +30,7 @@ static const char *RmgrNames[RM_MAX_ID + 1] = {
 };
 
 static void extractPageInfo(XLogReaderState *record);
+static bool getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime);
 
 static int	xlogreadfd = -1;
 static XLogSegNo xlogreadsegno = -1;
@@ -175,7 +176,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 static void
 extractPageInfo(XLogReaderState *record)
 {
-	int			block_id;
+	uint8		block_id;
 	RmgrId		rmid = XLogRecGetRmid(record);
 	uint8		info = XLogRecGetInfo(record);
 	uint8		rminfo = info & ~XLR_INFO_MASK;
@@ -240,3 +241,39 @@ extractPageInfo(XLogReaderState *record)
 		process_block_change(forknum, rnode, blkno);
 	}
 }
+
+/*
+ * Extract timestamp from WAL record.
+ *
+ * If the record contains a timestamp, returns true, and saves the timestamp
+ * in *recordXtime. If the record type has no timestamp, returns false.
+ * Currently, only transaction commit/abort records and restore points contain
+ * timestamps.
+ */
+static bool
+getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime)
+{
+	uint8		info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
+	uint8		xact_info = info & XLOG_XACT_OPMASK;
+	uint8		rmid = XLogRecGetRmid(record);
+
+	if (rmid == RM_XLOG_ID && info == XLOG_RESTORE_POINT)
+	{
+		*recordXtime = ((xl_restore_point *) XLogRecGetData(record))->rp_time;
+		return true;
+	}
+	if (rmid == RM_XACT_ID && (xact_info == XLOG_XACT_COMMIT ||
+							   xact_info == XLOG_XACT_COMMIT_PREPARED))
+	{
+		*recordXtime = ((xl_xact_commit *) XLogRecGetData(record))->xact_time;
+		return true;
+	}
+	if (rmid == RM_XACT_ID && (xact_info == XLOG_XACT_ABORT ||
+							   xact_info == XLOG_XACT_ABORT_PREPARED))
+	{
+		*recordXtime = ((xl_xact_abort *) XLogRecGetData(record))->xact_time;
+		return true;
+	}
+	return false;
+}
+

pg_probackup.c

@@ -206,18 +206,25 @@ main(int argc, char *argv[])
 		if (res != 0)
 			return res;
 
-		do_validate(current.start_time);
+		do_validate_last();
 	}
 	else if (pg_strcasecmp(cmd, "restore") == 0)
-		return do_restore(backup_id, target_time, target_xid,
-					target_inclusive, target_tli);
+		return do_restore(backup_id,
+						  target_time,
+						  target_xid,
+						  target_inclusive,
+						  target_tli);
 	else if (pg_strcasecmp(cmd, "show") == 0)
 		return do_show(backup_id);
 	else if (pg_strcasecmp(cmd, "validate") == 0)
 	{
 		if (backup_id == 0)
 			elog(ERROR, "you must specify backup-ID for this command");
-		return do_validate(backup_id);
+		return do_validate(backup_id,
+						   target_time,
+						   target_xid,
+						   target_inclusive,
+						   target_tli);
 	}
 	else if (pg_strcasecmp(cmd, "delete") == 0)
 		return do_delete(backup_id);

pg_probackup.h

@@ -221,6 +221,17 @@ extern int do_restore(time_t backup_id,
 					  const char *target_xid,
 					  const char *target_inclusive,
 					  TimeLineID target_tli);
+extern bool satisfy_timeline(const parray *timelines, const pgBackup *backup);
+extern bool satisfy_recovery_target(const pgBackup *backup,
+									const pgRecoveryTarget *rt);
+extern TimeLineID get_fullbackup_timeline(parray *backups,
+										  const pgRecoveryTarget *rt);
+extern TimeLineID findNewestTimeLine(TimeLineID startTLI);
+extern parray * readTimeLineHistory(TimeLineID targetTLI);
+extern pgRecoveryTarget *checkIfCreateRecoveryConf(
+	const char *target_time,
+	const char *target_xid,
+	const char *target_inclusive);
 
 /* in init.c */
 extern int do_init(void);
@@ -240,7 +251,12 @@ extern char *slurpFile(const char *datadir,
 					   bool safe);
 
 /* in validate.c */
-extern int do_validate(time_t backup_id);
+extern int do_validate(time_t backup_id,
+					   const char *target_time,
+					   const char *target_xid,
+					   const char *target_inclusive,
+					   TimeLineID target_tli);
+extern void do_validate_last(void);
 extern void pgBackupValidate(pgBackup *backup,
 							 bool size_only,
 							 bool for_get_timeline);
@@ -316,4 +332,46 @@ extern uint64 get_system_identifier(bool safe);
 /* in status.c */
 extern bool is_pg_running(void);
 
+/* some from access/xact.h */
+/*
+ * XLOG allows to store some information in high 4 bits of log record xl_info
+ * field. We use 3 for the opcode, and one about an optional flag variable.
+ */
+#define XLOG_XACT_COMMIT			0x00
+#define XLOG_XACT_PREPARE			0x10
+#define XLOG_XACT_ABORT				0x20
+#define XLOG_XACT_COMMIT_PREPARED	0x30
+#define XLOG_XACT_ABORT_PREPARED	0x40
+#define XLOG_XACT_ASSIGNMENT		0x50
+/* free opcode 0x60 */
+/* free opcode 0x70 */
+
+/* mask for filtering opcodes out of xl_info */
+#define XLOG_XACT_OPMASK			0x70
+
+typedef struct xl_xact_commit
+{
+	TimestampTz xact_time;		/* time of commit */
+
+	/* xl_xact_xinfo follows if XLOG_XACT_HAS_INFO */
+	/* xl_xact_dbinfo follows if XINFO_HAS_DBINFO */
+	/* xl_xact_subxacts follows if XINFO_HAS_SUBXACT */
+	/* xl_xact_relfilenodes follows if XINFO_HAS_RELFILENODES */
+	/* xl_xact_invals follows if XINFO_HAS_INVALS */
+	/* xl_xact_twophase follows if XINFO_HAS_TWOPHASE */
+	/* xl_xact_origin follows if XINFO_HAS_ORIGIN, stored unaligned! */
+} xl_xact_commit;
+
+typedef struct xl_xact_abort
+{
+	TimestampTz xact_time;		/* time of abort */
+
+	/* xl_xact_xinfo follows if XLOG_XACT_HAS_INFO */
+	/* No db_info required */
+	/* xl_xact_subxacts follows if HAS_SUBXACT */
+	/* xl_xact_relfilenodes follows if HAS_RELFILENODES */
+	/* No invalidation messages needed. */
+	/* xl_xact_twophase follows if XINFO_HAS_TWOPHASE */
+} xl_xact_abort;
+
 #endif /* PG_PROBACKUP_H */

restore.c

@@ -29,22 +29,13 @@ static void create_recovery_conf(time_t backup_id,
 								 const char *target_xid,
 								 const char *target_inclusive,
 								 TimeLineID target_tli);
-static pgRecoveryTarget *checkIfCreateRecoveryConf(const char *target_time,
-								 const char *target_xid,
-								 const char *target_inclusive);
-static parray * readTimeLineHistory(TimeLineID targetTLI);
-static bool satisfy_timeline(const parray *timelines, const pgBackup *backup);
-static bool satisfy_recovery_target(const pgBackup *backup,
-									const pgRecoveryTarget *rt);
-static TimeLineID get_fullbackup_timeline(parray *backups,
-										  const pgRecoveryTarget *rt);
 static void print_backup_lsn(const pgBackup *backup);
 static void search_next_wal(const char *path,
 							XLogRecPtr *need_lsn,
 							parray *timelines);
 static void restore_files(void *arg);
 
-TimeLineID findNewestTimeLine(TimeLineID startTLI);
+
 bool existsTimeLineHistory(TimeLineID probeTLI);
 
 
@@ -63,9 +54,11 @@ do_restore(time_t backup_id,
 	TimeLineID	backup_tli;
 	TimeLineID	newest_tli;
 	parray *backups;
-	pgBackup *base_backup = NULL;
+
 	parray *files;
 	parray *timelines;
+	pgBackup *base_backup = NULL;
+	pgBackup *dest_backup = NULL;
 	pgRecoveryTarget *rt = NULL;
 	XLogRecPtr need_lsn;
 	bool backup_id_found = false;
@@ -126,15 +119,25 @@ do_restore(time_t backup_id,
 			continue;
 
 		if (backup_id == base_backup->start_time &&
-			base_backup->status == BACKUP_STATUS_OK
-		)
+			base_backup->status == BACKUP_STATUS_OK)
+		{
 			backup_id_found = true;
+			dest_backup = base_backup;
+		}
 
 		if (backup_id == base_backup->start_time &&
 			base_backup->status != BACKUP_STATUS_OK
 		)
 			elog(ERROR, "given backup %s is %s", base36enc(backup_id), status2str(base_backup->status));
 
+		if (dest_backup != NULL &&
+			base_backup->backup_mode == BACKUP_MODE_FULL &&
+			base_backup->status != BACKUP_STATUS_OK)
+			elog(ERROR, "base backup %s for given backup %s is %s",
+				 base36enc(base_backup->start_time),
+				 base36enc(dest_backup->start_time),
+				 status2str(base_backup->status));
+
 		if (base_backup->backup_mode < BACKUP_MODE_FULL ||
 			base_backup->status != BACKUP_STATUS_OK)
 			continue;
@@ -522,7 +525,7 @@ create_recovery_conf(time_t backup_id,
  * specified timeline ID.
  * based on readTimeLineHistory() in xlog.c
  */
-static parray *
+parray *
 readTimeLineHistory(TimeLineID targetTLI)
 {
 	parray	   *result;
@@ -632,7 +635,7 @@ readTimeLineHistory(TimeLineID targetTLI)
 	return result;
 }
 
-static bool
+bool
 satisfy_recovery_target(const pgBackup *backup, const pgRecoveryTarget *rt)
 {
 	if (rt->xid_specified)
@@ -644,7 +647,7 @@ satisfy_recovery_target(const pgBackup *backup, const pgRecoveryTarget *rt)
 	return true;
 }
 
-static bool
+bool
 satisfy_timeline(const parray *timelines, const pgBackup *backup)
 {
 	int i;
@@ -659,7 +662,7 @@ satisfy_timeline(const parray *timelines, const pgBackup *backup)
 }
 
 /* get TLI of the latest full backup */
-static TimeLineID
+TimeLineID
 get_fullbackup_timeline(parray *backups, const pgRecoveryTarget *rt)
 {
 	int			i;
@@ -769,7 +772,7 @@ search_next_wal(const char *path, XLogRecPtr *need_lsn, parray *timelines)
 	}
 }
 
-static pgRecoveryTarget *
+pgRecoveryTarget *
 checkIfCreateRecoveryConf(const char *target_time,
                    const char *target_xid,
                    const char *target_inclusive)