PGPRO-1449: Add MERGE command: merge PAGE with FULL backup

Author: za-arthur

Makefile

@@ -5,7 +5,7 @@ OBJS = src/backup.o src/catalog.o src/configure.o src/data.o \
 	src/util.o src/validate.o src/datapagemap.o src/parsexlog.o \
 	src/xlogreader.o src/streamutil.o src/receivelog.o \
 	src/archive.o src/utils/parray.o src/utils/pgut.o src/utils/logger.o \
-	src/utils/json.o src/utils/thread.o
+	src/utils/json.o src/utils/thread.o src/merge.o
 
 EXTRA_CLEAN = src/datapagemap.c src/datapagemap.h src/xlogreader.c \
 	src/receivelog.c src/receivelog.h src/streamutil.c src/streamutil.h src/logging.h

src/backup.c

@@ -104,7 +104,6 @@ static int checkpoint_timeout(void);
 
 //static void backup_list_file(parray *files, const char *root, )
 static void parse_backup_filelist_filenames(parray *files, const char *root);
-static void write_backup_file_list(parray *files, const char *root);
 static void wait_wal_lsn(XLogRecPtr lsn, bool wait_prev_segment);
 static void wait_replica_wal_lsn(XLogRecPtr lsn, bool is_start_backup);
 static void make_pagemap_from_ptrack(parray *files);
@@ -515,8 +514,6 @@ do_backup_instance(void)
 
 		/* get list of backups already taken */
 		backup_list = catalog_get_backup_list(INVALID_BACKUP_ID);
-		if (backup_list == NULL)
-			elog(ERROR, "Failed to get backup list.");
 
 		prev_backup = catalog_get_last_data_backup(backup_list, current.tli);
 		if (prev_backup == NULL)
@@ -697,8 +694,11 @@ do_backup_instance(void)
 		pg_atomic_clear_flag(&file->lock);
 	}
 
-	/* sort by size for load balancing */
+	/* Sort by size for load balancing */
 	parray_qsort(backup_files_list, pgFileCompareSize);
+	/* Sort the array for binary search */
+	if (prev_backup_filelist)
+		parray_qsort(prev_backup_filelist, pgFileComparePath);
 
 	/* init thread args with own file lists */
 	threads = (pthread_t *) palloc(sizeof(pthread_t) * num_threads);
@@ -788,7 +788,7 @@ do_backup_instance(void)
 	}
 
 	/* Print the list of files to backup catalog */
-	write_backup_file_list(backup_files_list, pgdata);
+	pgBackupWriteFileList(&current, backup_files_list, pgdata);
 
 	/* Compute summary of size of regular files in the backup */
 	for (i = 0; i < parray_num(backup_files_list); i++)
@@ -2076,35 +2076,32 @@ backup_files(void *arg)
 			/* Check that file exist in previous backup */
 			if (current.backup_mode != BACKUP_MODE_FULL)
 			{
-				int			p;
 				char	   *relative;
-				int			n_prev_files = parray_num(arguments->prev_filelist);
+				pgFile		key;
+				pgFile	  **prev_file;
 
 				relative = GetRelativePath(file->path, arguments->from_root);
-				for (p = 0; p < n_prev_files; p++)
-				{
-					pgFile	   *prev_file;
-
-					prev_file = (pgFile *) parray_get(arguments->prev_filelist, p);
+				key.path = relative;
 
-					if (strcmp(relative, prev_file->path) == 0)
-					{
-						/* File exists in previous backup */
-						file->exists_in_prev = true;
-						// elog(VERBOSE, "File exists at the time of previous backup %s", relative);
-						break;
-					}
-				}
+				prev_file = (pgFile **) parray_bsearch(arguments->prev_filelist,
+													   &key, pgFileComparePath);
+				if (prev_file)
+					/* File exists in previous backup */
+					file->exists_in_prev = true;
 			}
 			/* copy the file into backup */
 			if (file->is_datafile && !file->is_cfs)
 			{
+				char		to_path[MAXPGPATH];
+
+				join_path_components(to_path, arguments->to_root,
+									 file->path + strlen(arguments->from_root) + 1);
+
 				/* backup block by block if datafile AND not compressed by cfs*/
-				if (!backup_data_file(arguments,
-									  arguments->from_root,
-									  arguments->to_root, file,
+				if (!backup_data_file(arguments, to_path, file,
 									  arguments->prev_start_lsn,
-									  current.backup_mode))
+									  current.backup_mode,
+									  compress_alg, compress_level))
 				{
 					file->write_size = BYTES_INVALID;
 					elog(VERBOSE, "File \"%s\" was not copied to backup", file->path);
@@ -2279,31 +2276,6 @@ set_cfs_datafiles(parray *files, const char *root, char *relative, size_t i)
 	free(cfs_tblspc_path);
 }
 
-
-/*
- * Output the list of files to backup catalog DATABASE_FILE_LIST
- */
-static void
-write_backup_file_list(parray *files, const char *root)
-{
-	FILE	   *fp;
-	char		path[MAXPGPATH];
-
-	pgBackupGetPath(&current, path, lengthof(path), DATABASE_FILE_LIST);
-
-	fp = fopen(path, "wt");
-	if (fp == NULL)
-		elog(ERROR, "cannot open file list \"%s\": %s", path,
-			strerror(errno));
-
-	print_file_list(fp, files, root);
-
-	if (fflush(fp) != 0 ||
-		fsync(fileno(fp)) != 0 ||
-		fclose(fp))
-		elog(ERROR, "cannot write file list \"%s\": %s", path, strerror(errno));
-}
-
 /*
  * Find pgfile by given rnode in the backup_files_list
  * and add given blkno to its pagemap.

src/catalog.c

@@ -325,6 +325,9 @@ catalog_get_backup_list(time_t requested_backup_id)
 	if (backups)
 		parray_walk(backups, pgBackupFree);
 	parray_free(backups);
+
+	elog(ERROR, "Failed to get backup list");
+
 	return NULL;
 }
 
@@ -389,7 +392,7 @@ pgBackupWriteControl(FILE *out, pgBackup *backup)
 			deparse_compress_alg(backup->compress_alg));
 	fprintf(out, "compress-level = %d\n", backup->compress_level);
 	fprintf(out, "from-replica = %s\n", backup->from_replica ? "true" : "false");
-	
+
 	fprintf(out, "\n#Compatibility\n");
 	fprintf(out, "block-size = %u\n", backup->block_size);
 	fprintf(out, "xlog-block-size = %u\n", backup->wal_block_size);
@@ -462,6 +465,30 @@ pgBackupWriteBackupControlFile(pgBackup *backup)
 	fclose(fp);
 }
 
+/*
+ * Output the list of files to backup catalog DATABASE_FILE_LIST
+ */
+void
+pgBackupWriteFileList(pgBackup *backup, parray *files, const char *root)
+{
+	FILE	   *fp;
+	char		path[MAXPGPATH];
+
+	pgBackupGetPath(backup, path, lengthof(path), DATABASE_FILE_LIST);
+
+	fp = fopen(path, "wt");
+	if (fp == NULL)
+		elog(ERROR, "cannot open file list \"%s\": %s", path,
+			strerror(errno));
+
+	print_file_list(fp, files, root);
+
+	if (fflush(fp) != 0 ||
+		fsync(fileno(fp)) != 0 ||
+		fclose(fp))
+		elog(ERROR, "cannot write file list \"%s\": %s", path, strerror(errno));
+}
+
 /*
  * Read BACKUP_CONTROL_FILE and create pgBackup.
  *  - Comment starts with ';'.
@@ -515,7 +542,7 @@ readBackupControlFile(const char *path)
 		return NULL;
 	}
 
-	pgBackup_init(backup);
+	pgBackupInit(backup);
 	parsed_options = pgut_readopt(path, options, WARNING);
 
 	if (parsed_options == 0)
@@ -566,10 +593,12 @@ readBackupControlFile(const char *path)
 	{
 		if (strcmp(status, "OK") == 0)
 			backup->status = BACKUP_STATUS_OK;
-		else if (strcmp(status, "RUNNING") == 0)
-			backup->status = BACKUP_STATUS_RUNNING;
 		else if (strcmp(status, "ERROR") == 0)
 			backup->status = BACKUP_STATUS_ERROR;
+		else if (strcmp(status, "RUNNING") == 0)
+			backup->status = BACKUP_STATUS_RUNNING;
+		else if (strcmp(status, "MERGING") == 0)
+			backup->status = BACKUP_STATUS_MERGING;
 		else if (strcmp(status, "DELETING") == 0)
 			backup->status = BACKUP_STATUS_DELETING;
 		else if (strcmp(status, "DELETED") == 0)
@@ -698,11 +727,63 @@ deparse_compress_alg(int alg)
 	return NULL;
 }
 
+/*
+ * Fill pgBackup struct with default values.
+ */
+void
+pgBackupInit(pgBackup *backup)
+{
+	backup->backup_id = INVALID_BACKUP_ID;
+	backup->backup_mode = BACKUP_MODE_INVALID;
+	backup->status = BACKUP_STATUS_INVALID;
+	backup->tli = 0;
+	backup->start_lsn = 0;
+	backup->stop_lsn = 0;
+	backup->start_time = (time_t) 0;
+	backup->end_time = (time_t) 0;
+	backup->recovery_xid = 0;
+	backup->recovery_time = (time_t) 0;
+
+	backup->data_bytes = BYTES_INVALID;
+	backup->wal_bytes = BYTES_INVALID;
+
+	backup->compress_alg = COMPRESS_ALG_DEFAULT;
+	backup->compress_level = COMPRESS_LEVEL_DEFAULT;
+
+	backup->block_size = BLCKSZ;
+	backup->wal_block_size = XLOG_BLCKSZ;
+	backup->checksum_version = 0;
+
+	backup->stream = false;
+	backup->from_replica = false;
+	backup->parent_backup = INVALID_BACKUP_ID;
+	backup->primary_conninfo = NULL;
+	backup->program_version[0] = '\0';
+	backup->server_version[0] = '\0';
+}
+
+/*
+ * Copy backup metadata from **src** into **dst**.
+ */
+void
+pgBackupCopy(pgBackup *dst, pgBackup *src)
+{
+	pfree(dst->primary_conninfo);
+
+	memcpy(dst, src, sizeof(pgBackup));
+
+	if (src->primary_conninfo)
+		dst->primary_conninfo = pstrdup(src->primary_conninfo);
+}
+
 /* free pgBackup object */
 void
 pgBackupFree(void *backup)
 {
-	free(backup);
+	pgBackup *b = (pgBackup *) backup;
+
+	pfree(b->primary_conninfo);
+	pfree(backup);
 }
 
 /* Compare two pgBackup with their IDs (start time) in ascending order */