Merge remote-tracking branch 'origin/master' into pgpro-1918

Author: za-arthur

src/backup.c

@@ -574,6 +574,9 @@ do_backup_instance(void)
 			strlen(" with pg_probackup"));
 	pg_start_backup(label, smooth_checkpoint, &current);
 
+	/* Update running backup meta with START LSN */
+	write_backup(&current);
+
 	pgBackupGetPath(&current, database_path, lengthof(database_path),
 					DATABASE_DIR);
 
@@ -760,6 +763,19 @@ do_backup_instance(void)
 	else
 		elog(ERROR, "Data files transferring failed");
 
+	/* Remove disappeared during backup files from backup_list */
+	for (i = 0; i < parray_num(backup_files_list); i++)
+	{
+		pgFile	   *tmp_file = (pgFile *) parray_get(backup_files_list, i);
+
+		if (tmp_file->write_size == FILE_NOT_FOUND)
+		{
+			pg_atomic_clear_flag(&tmp_file->lock);
+			pgFileFree(tmp_file);
+			parray_remove(backup_files_list, i);
+		}
+	}
+
 	/* clean previous backup file list */
 	if (prev_backup_filelist)
 	{
@@ -2235,7 +2251,7 @@ backup_files(void *arg)
 				 * If file is not found, this is not en error.
 				 * It could have been deleted by concurrent postgres transaction.
 				 */
-				file->write_size = BYTES_INVALID;
+				file->write_size = FILE_NOT_FOUND;
 				elog(LOG, "File \"%s\" is not found", file->path);
 				continue;
 			}
@@ -2285,7 +2301,9 @@ backup_files(void *arg)
 									  instance_config.compress_alg,
 									  instance_config.compress_level))
 				{
-					file->write_size = BYTES_INVALID;
+					/* disappeared file not to be confused with 'not changed' */
+					if (file->write_size != FILE_NOT_FOUND)
+						file->write_size = BYTES_INVALID;
 					elog(VERBOSE, "File \"%s\" was not copied to backup", file->path);
 					continue;
 				}
@@ -2309,7 +2327,9 @@ backup_files(void *arg)
 				if (skip ||
 					!copy_file(arguments->from_root, arguments->to_root, file))
 				{
-					file->write_size = BYTES_INVALID;
+					/* disappeared file not to be confused with 'not changed' */
+					if (file->write_size != FILE_NOT_FOUND)
+						file->write_size = BYTES_INVALID;
 					elog(VERBOSE, "File \"%s\" was not copied to backup",
 						 file->path);
 					continue;

src/catalog.c

@@ -1021,6 +1021,9 @@ is_parent(time_t parent_backup_time, pgBackup *child_backup, bool inclusive)
 	if (!child_backup)
 		elog(ERROR, "Target backup cannot be NULL");
 
+	if (inclusive && child_backup->start_time == parent_backup_time)
+		return true;
+
 	while (child_backup->parent_backup_link &&
 			child_backup->parent_backup != parent_backup_time)
 	{
@@ -1030,8 +1033,8 @@ is_parent(time_t parent_backup_time, pgBackup *child_backup, bool inclusive)
 	if (child_backup->parent_backup == parent_backup_time)
 		return true;
 
-	if (inclusive && child_backup->start_time == parent_backup_time)
-		return true;
+	//if (inclusive && child_backup->start_time == parent_backup_time)
+	//	return true;
 
 	return false;
 }

src/data.c

@@ -564,6 +564,7 @@ backup_data_file(backup_files_arg* arguments,
 		if (errno == ENOENT)
 		{
 			elog(LOG, "File \"%s\" is not found", file->path);
+			file->write_size = FILE_NOT_FOUND;
 			return false;
 		}
 
@@ -946,7 +947,11 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
 
 		/* maybe deleted, it's not error */
 		if (errno == ENOENT)
+		{
+			elog(LOG, "File \"%s\" is not found", file->path);
+			file->write_size = FILE_NOT_FOUND;
 			return false;
+		}
 
 		elog(ERROR, "cannot open source file \"%s\": %s", file->path,
 			 strerror(errno));

src/dir.c

@@ -630,6 +630,14 @@ dir_check_file(const char *root, pgFile *file)
 			}
 			else
 			{
+               /*
+                * snapfs files:
+                * RELFILENODE.BLOCKNO.snapmap.SNAPID
+                * RELFILENODE.BLOCKNO.snap.SNAPID
+                */
+               if (strstr(file->name, "snap") != NULL)
+                       return true;
+
 				len = strlen(file->name);
 				/* reloid.cfm */
 				if (len > 3 && strcmp(file->name + len - 3, "cfm") == 0)
@@ -1047,11 +1055,11 @@ create_data_directories(const char *data_dir, const char *backup_dir,
 				}
 
 				if (link_sep)
-					elog(LOG, "create directory \"%s\" and symbolic link \"%.*s\"",
+					elog(VERBOSE, "create directory \"%s\" and symbolic link \"%.*s\"",
 						 linked_path,
 						 (int) (link_sep -  relative_ptr), relative_ptr);
 				else
-					elog(LOG, "create directory \"%s\" and symbolic link \"%s\"",
+					elog(VERBOSE, "create directory \"%s\" and symbolic link \"%s\"",
 						 linked_path, relative_ptr);
 
 				/* Firstly, create linked directory */
@@ -1082,7 +1090,7 @@ create_data_directories(const char *data_dir, const char *backup_dir,
 		}
 
 create_directory:
-		elog(LOG, "create directory \"%s\"", relative_ptr);
+		elog(VERBOSE, "create directory \"%s\"", relative_ptr);
 
 		/* This is not symlink, create directory */
 		join_path_components(to_path, data_dir, relative_ptr);

src/parsexlog.c

@@ -93,6 +93,9 @@ typedef struct XLogPageReadPrivate
 	TimeLineID	tli;
 	uint32		xlog_seg_size;
 
+	char		page_buf[XLOG_BLCKSZ];
+	uint32		prev_page_off;
+
 	bool		manual_switch;
 	bool		need_switch;
 
@@ -1037,6 +1040,17 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 	 */
 	Assert(private_data->xlogexists);
 
+	/*
+	 * Do not read same page read earlier from the file, read it from the buffer
+	 */
+	if (private_data->prev_page_off != 0 &&
+		private_data->prev_page_off == targetPageOff)
+	{
+		memcpy(readBuf, private_data->page_buf, XLOG_BLCKSZ);
+		*pageTLI = private_data->tli;
+		return XLOG_BLCKSZ;
+	}
+
 	/* Read the requested page */
 	if (private_data->xlogfile != -1)
 	{
@@ -1060,23 +1074,25 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 		if (gzseek(private_data->gz_xlogfile, (z_off_t) targetPageOff, SEEK_SET) == -1)
 		{
 			elog(WARNING, "Thread [%d]: Could not seek in compressed WAL segment \"%s\": %s",
-				 private_data->thread_num,
-				 private_data->gz_xlogpath,
-				 get_gz_error(private_data->gz_xlogfile));
+				private_data->thread_num,
+				private_data->gz_xlogpath,
+				get_gz_error(private_data->gz_xlogfile));
 			return -1;
 		}
 
 		if (gzread(private_data->gz_xlogfile, readBuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
 		{
 			elog(WARNING, "Thread [%d]: Could not read from compressed WAL segment \"%s\": %s",
-				 private_data->thread_num,
-				 private_data->gz_xlogpath,
-				 get_gz_error(private_data->gz_xlogfile));
+				private_data->thread_num,
+				private_data->gz_xlogpath,
+				get_gz_error(private_data->gz_xlogfile));
 			return -1;
 		}
 	}
 #endif
 
+	memcpy(private_data->page_buf, readBuf, XLOG_BLCKSZ);
+	private_data->prev_page_off = targetPageOff;
 	*pageTLI = private_data->tli;
 	return XLOG_BLCKSZ;
 }
@@ -1133,6 +1149,7 @@ CleanupXLogPageRead(XLogReaderState *xlogreader)
 		private_data->gz_xlogfile = NULL;
 	}
 #endif
+	private_data->prev_page_off = 0;
 	private_data->xlogexists = false;
 }
 

src/pg_probackup.h

@@ -165,7 +165,8 @@ typedef enum ShowFormat
 
 /* special values of pgBackup fields */
 #define INVALID_BACKUP_ID	0    /* backup ID is not provided by user */
-#define BYTES_INVALID		(-1)
+#define BYTES_INVALID		(-1) /* file didn`t changed since previous backup, DELTA backup do not rely on it */
+#define FILE_NOT_FOUND		(-2) /* file disappeared during backup */
 #define BLOCKNUM_INVALID	(-1)
 
 /*

src/restore.c

@@ -57,6 +57,7 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 	int			base_full_backup_index = 0;
 	int			corrupted_backup_index = 0;
 	char	   *action = is_restore ? "Restore":"Validate";
+	parray	   *parent_chain = NULL;
 
 	if (is_restore)
 	{
@@ -283,34 +284,54 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 	if (is_restore)
 		check_tablespace_mapping(dest_backup);
 
+	/* At this point we are sure that parent chain is whole
+	 * so we can build separate array, containing all needed backups,
+	 * to simplify validation and restore
+	 */
+	parent_chain = parray_new();
+
+	/* Take every backup that is a child of base_backup AND parent of dest_backup
+	 * including base_backup and dest_backup
+	 */
+	for (i = base_full_backup_index; i >= dest_backup_index; i--)
+	{
+		tmp_backup = (pgBackup *) parray_get(backups, i);
+
+		if (is_parent(base_full_backup->start_time, tmp_backup, true) &&
+					is_parent(tmp_backup->start_time, dest_backup, true))
+		{
+			parray_append(parent_chain, tmp_backup);
+		}
+	}
+
+	/* for validation or restore with enabled validation */
 	if (!is_restore || !rt->restore_no_validate)
 	{
 		if (dest_backup->backup_mode != BACKUP_MODE_FULL)
 			elog(INFO, "Validating parents for backup %s", base36enc(dest_backup->start_time));
 
 		/*
 		 * Validate backups from base_full_backup to dest_backup.
-		 * At this point we are sure that parent chain is intact.
 		 */
-		for (i = base_full_backup_index; i >= dest_backup_index; i--)
+		for (i = 0; i < parray_num(parent_chain); i++)
 		{
-			tmp_backup = (pgBackup *) parray_get(backups, i);
+			tmp_backup = (pgBackup *) parray_get(parent_chain, i);
 
-			if (is_parent(base_full_backup->start_time, tmp_backup, true))
+			lock_backup(tmp_backup);
+			pgBackupValidate(tmp_backup);
+			/* Maybe we should be more paranoid and check for !BACKUP_STATUS_OK? */
+			if (tmp_backup->status == BACKUP_STATUS_CORRUPT)
 			{
-				lock_backup(tmp_backup);
-				pgBackupValidate(tmp_backup);
-				/* Maybe we should be more paranoid and check for !BACKUP_STATUS_OK? */
-				if (tmp_backup->status == BACKUP_STATUS_CORRUPT)
-				{
-					corrupted_backup = tmp_backup;
-					corrupted_backup_index = i;
-					break;
-				}
-				/* We do not validate WAL files of intermediate backups
-				 * It`s done to speed up restore
+				corrupted_backup = tmp_backup;
+				/* we need corrupted backup index from 'backups' not parent_chain
+				 * so we can properly orphanize all its descendants
 				 */
+				corrupted_backup_index = get_backup_index_number(backups, corrupted_backup);
+				break;
 			}
+			/* We do not validate WAL files of intermediate backups
+			 * It`s done to speed up restore
+			 */
 		}
 
 		/* There is no point in wal validation of corrupted backups */
@@ -353,7 +374,6 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 		}
 	}
 
-	// TODO: rewrite restore to use parent_chain
 	/*
 	 * If dest backup is corrupted or was orphaned in previous check
 	 * produce corresponding error message
@@ -374,13 +394,12 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 				base36enc(dest_backup->start_time), status2str(dest_backup->status));
 
 	/* We ensured that all backups are valid, now restore if required
-	 * TODO: use parent_link
 	 */
 	if (is_restore)
 	{
-		for (i = base_full_backup_index; i >= dest_backup_index; i--)
+		for (i = 0; i < parray_num(parent_chain); i++)
 		{
-			pgBackup   *backup = (pgBackup *) parray_get(backups, i);
+			pgBackup   *backup = (pgBackup *) parray_get(parent_chain, i);
 
 			if (rt->lsn_specified && parse_server_version(backup->server_version) < 100000)
 				elog(ERROR, "Backup %s was created for version %s which doesn't support recovery_target_lsn",
@@ -410,6 +429,7 @@ do_restore_or_validate(time_t target_backup_id, pgRecoveryTarget *rt,
 	/* cleanup */
 	parray_walk(backups, pgBackupFree);
 	parray_free(backups);
+	parray_free(parent_chain);
 
 	elog(INFO, "%s of backup %s completed.",
 		 action, base36enc(dest_backup->start_time));
@@ -485,6 +505,7 @@ restore_backup(pgBackup *backup)
 		/* By default there are some error */
 		threads_args[i].ret = 1;
 
+		/* Useless message TODO: rewrite */
 		elog(LOG, "Start thread for num:%zu", parray_num(files));
 
 		pthread_create(&threads[i], NULL, restore_files, arg);

src/show.c

@@ -325,6 +325,7 @@ show_instance_plain(parray *backup_list, bool show_name)
 	uint32		widths[SHOW_FIELDS_COUNT];
 	uint32		widths_sum = 0;
 	ShowBackendRow *rows;
+	time_t current_time = time(NULL);
 
 	for (i = 0; i < SHOW_FIELDS_COUNT; i++)
 		widths[i] = strlen(names[i]);
@@ -384,7 +385,11 @@ show_instance_plain(parray *backup_list, bool show_name)
 		cur++;
 
 		/* Time */
-		if (backup->end_time != (time_t) 0)
+		if (backup->status == BACKUP_STATUS_RUNNING)
+			snprintf(row->duration, lengthof(row->duration), "%.*lfs", 0,
+					 difftime(current_time, backup->start_time));
+
+		else if (backup->end_time != (time_t) 0)
 			snprintf(row->duration, lengthof(row->duration), "%.*lfs", 0,
 					 difftime(backup->end_time, backup->start_time));
 		else

tests/__init__.py

@@ -8,7 +8,7 @@
     ptrack_vacuum_full, ptrack_vacuum_truncate, pgpro560, pgpro589, \
     false_positive, replica, compression, page, ptrack, archive, \
     exclude, cfs_backup, cfs_restore, cfs_validate_backup, auth_test, \
-    time_stamp, snapfs, logging, locking
+    time_stamp, snapfs, logging, locking, remote
 
 
 def load_tests(loader, tests, pattern):
@@ -44,6 +44,7 @@ def load_tests(loader, tests, pattern):
     suite.addTests(loader.loadTestsFromModule(ptrack_vacuum_bits_visibility))
     suite.addTests(loader.loadTestsFromModule(ptrack_vacuum_full))
     suite.addTests(loader.loadTestsFromModule(ptrack_vacuum_truncate))
+    suite.addTests(loader.loadTestsFromModule(remote))
     suite.addTests(loader.loadTestsFromModule(replica))
     suite.addTests(loader.loadTestsFromModule(restore_test))
     suite.addTests(loader.loadTestsFromModule(retention_test))