Implement incremental backup of compressed files, both page and ptrack do the same. Adds two columns to the list of files.

backup.c

@@ -1102,6 +1102,75 @@ backup_cleanup(bool fatal, void *userdata)
 	}
 }
 
+/* Count bytes in file */
+static long
+file_size(const char *file)
+{
+	long		r;
+	FILE	   *f = fopen(file, "r");
+
+	if (!f)
+	{
+		elog(ERROR, "pg_probackup: could not open file \"%s\" for reading: %s\n",
+				file, strerror(errno));
+		return -1;
+	}
+	fseek(f, 0, SEEK_END);
+	r = ftell(f);
+	fclose(f);
+	return r;
+}
+
+/*
+ * Find corresponding file in previous backup.
+ * Compare generations and return true if we don't need full copy
+ * of the file, but just part of it.
+ *
+ * skip_size - size of the file in previous backup. We can skip it
+ *			   and copy just remaining part of the file.
+ */
+bool
+backup_compressed_file_partially(pgFile *file, void *arg, size_t *skip_size)
+{
+	bool result = false;
+	pgFile *prev_file = NULL;
+	size_t current_file_size;
+	backup_files_args *arguments = (backup_files_args *) arg;
+
+	if (arguments->prev_files)
+	{
+		pgFile **p = (pgFile **) parray_bsearch(arguments->prev_files,
+												file, pgFileComparePath);
+		if (p)
+			prev_file = *p;
+
+		/* If file's gc generation has changed since last backup, just copy it*/
+		if (prev_file && prev_file->generation == file->generation)
+		{
+			current_file_size = file_size(file->path);
+
+			if (prev_file->write_size == BYTES_INVALID)
+				return false;
+
+			*skip_size = prev_file->write_size;
+
+			if (current_file_size >= prev_file->write_size)
+			{
+				elog(LOG, "Backup file %s partially: prev_size %lu, current_size  %lu",
+					 file->path, prev_file->write_size, current_file_size);
+				result = true;
+			}
+			else
+				elog(ERROR, "Something is wrong with %s. current_file_size %lu, prev %lu",
+					file->path, current_file_size, prev_file->write_size);
+		}
+		else
+			elog(LOG, "Copy full %s.", file->path);
+	}
+
+	return result;
+}
+
 /*
  * Take differential backup at page level.
  */
@@ -1200,9 +1269,47 @@ backup_files(void *arg)
 			}
 
 			/* copy the file into backup */
-			if (!(file->is_datafile
+			if (file->is_datafile)
-					? backup_data_file(arguments->from_root, arguments->to_root, file, arguments->lsn)
+			{
-					: copy_file(arguments->from_root, arguments->to_root, file)))
+				if (!backup_data_file(arguments->from_root,
+									  arguments->to_root, file,
+									  arguments->lsn))
+				{
+					/* record as skipped file in file_xxx.txt */
+					file->write_size = BYTES_INVALID;
+					elog(LOG, "skip");
+					continue;
+				}
+			}
+			else if (is_compressed_data_file(file))
+			{
+				size_t skip_size = 0;
+				if (backup_compressed_file_partially(file, arguments, &skip_size))
+				{
+					/* backup cfs segment partly */
+					if (!copy_file_partly(arguments->from_root,
+							   arguments->to_root,
+							   file, skip_size))
+					{
+						/* record as skipped file in file_xxx.txt */
+						file->write_size = BYTES_INVALID;
+						elog(LOG, "skip");
+						continue;
+					}
+				}
+				else if (!copy_file(arguments->from_root,
+							   arguments->to_root,
+							   file))
+				{
+					/* record as skipped file in file_xxx.txt */
+					file->write_size = BYTES_INVALID;
+					elog(LOG, "skip");
+					continue;
+				}
+			}
+			else if (!copy_file(arguments->from_root,
+							   arguments->to_root,
+							   file))
 			{
 				/* record as skipped file in file_xxx.txt */
 				file->write_size = BYTES_INVALID;
@@ -1251,14 +1358,14 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 		relative = file->path + strlen(root) + 1;
 		if (is_pgdata &&
 			!path_is_prefix_of_path("base", relative) &&
-			/*!path_is_prefix_of_path("global", relative) &&*/
+			/*!path_is_prefix_of_path("global", relative) &&*/ //TODO What's wrong with this line?
 			!path_is_prefix_of_path("pg_tblspc", relative))
 			continue;
 
 		/* Get file name from path */
 		fname = last_dir_separator(relative);
 
-		/* Remove temp tables */
+		/* Remove temp tables from the list */
 		if (fname[0] == 't' && isdigit(fname[1]))
 		{
 			pgFileFree(file);
@@ -1268,7 +1375,7 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 		}
 
 		path_len = strlen(file->path);
-		/* Get link ptrack file to realations files */
+		/* Get link ptrack file to relations files */
 		if (path_len > 6 && strncmp(file->path+(path_len-6), "ptrack", 6) == 0)
 		{
 			pgFile *search_file;
@@ -1277,12 +1384,15 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 			while(true) {
 				pgFile tmp_file;
 				tmp_file.path = pg_strdup(file->path);
-				/* I hope segno not more than 999999 */
+
+				/* Segno fits into 6 digits since it is not more than 4000 */
 				if (segno > 0)
 					sprintf(tmp_file.path+path_len-7, ".%d", segno);
 				else
 					tmp_file.path[path_len-7] = '\0';
+
 				pre_search_file = (pgFile **) parray_bsearch(list_file, &tmp_file, pgFileComparePath);
+
 				if (pre_search_file != NULL)
 				{
 					search_file = *pre_search_file;
@@ -1296,6 +1406,7 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 				segno++;
 			}
 
+			/* Remove ptrack file itself from backup list */
 			pgFileFree(file);
 			parray_remove(list_file, i);
 			i--;
@@ -1305,9 +1416,9 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 		/* compress map file it is not data file */
 		if (path_len > 4 && strncmp(file->path+(path_len-4), ".cfm", 4) == 0)
 		{
-			if (current.backup_mode == BACKUP_MODE_DIFF_PTRACK ||
+			if (current.backup_mode == BACKUP_MODE_DIFF_PAGE)
-				current.backup_mode == BACKUP_MODE_DIFF_PAGE)
+				elog(ERROR, "You can't use PAGE mode backup with compressed tablespace.\n"
-				elog(ERROR, "You can't use incremental backup with compress tablespace");
+							"Try FULL or PTRACK mode instead.");
 			continue;
 		}
 
@@ -1355,11 +1466,34 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
 			pgFile tmp_file;
 			tmp_file.path = pg_strdup(file->path);
 			tmp_file.path[path_len-4] = '\0';
-			pre_search_file = (pgFile **) parray_bsearch(list_file, &tmp_file, pgFileComparePath);
+			pre_search_file = (pgFile **) parray_bsearch(list_file,
+														 &tmp_file, pgFileComparePath);
 			if (pre_search_file != NULL)
 			{
-				(*pre_search_file)->is_datafile = false;
+				FileMap* map;
+				int md = open(file->path, O_RDWR|PG_BINARY, 0);
+				if (md < 0)
+					elog(ERROR, "add_files(). cannot open cfm file '%s'", file->path);
+
+				map = cfs_mmap(md);
+				if (map == MAP_FAILED)
+				{
+					elog(LOG, "add_files(). cfs_compression_ration failed to map file %s: %m", file->path);
+					close(md);
+					break;
 				}
+
+				(*pre_search_file)->generation = map->generation;
+				(*pre_search_file)->is_datafile = false;
+
+				if (cfs_munmap(map) < 0)
+					elog(LOG, "add_files(). CFS failed to unmap file %s: %m", file->path);
+				if (close(md) < 0)
+					elog(LOG, "add_files(). CFS failed to close file %s: %m", file->path);
+			}
+			else
+				elog(ERROR, "corresponding segment '%s' is not found", tmp_file.path);
+
 			pg_free(tmp_file.path);
 		}
 	}
@@ -1633,3 +1767,34 @@ StreamLog(void *arg)
 	PQfinish(conn);
 	conn = NULL;
 }
+
+
+FileMap* cfs_mmap(int md)
+{
+	FileMap* map;
+#ifdef WIN32
+    HANDLE mh = CreateFileMapping(_get_osfhandle(md), NULL, PAGE_READWRITE,
+								  0, (DWORD)sizeof(FileMap), NULL);
+    if (mh == NULL)
+        return (FileMap*)MAP_FAILED;
+
+    map = (FileMap*)MapViewOfFile(mh, FILE_MAP_ALL_ACCESS, 0, 0, 0);
+	CloseHandle(mh);
+    if (map == NULL)
+        return (FileMap*)MAP_FAILED;
+
+#else
+	map = (FileMap*)mmap(NULL, sizeof(FileMap),
+						 PROT_WRITE | PROT_READ, MAP_SHARED, md, 0);
+#endif
+	return map;
+}
+
+int cfs_munmap(FileMap* map)
+{
+#ifdef WIN32
+	return UnmapViewOfFile(map) ? 0 : -1;
+#else
+	return munmap(map, sizeof(FileMap));
+#endif
+}

data.c

@@ -113,7 +113,6 @@ backup_data_file(const char *from_root, const char *to_root,
 	/* confirm server version */
 	check_server_version();
 
-
 	/*
 	 * Read each page and write the page excluding hole. If it has been
 	 * determined that the page can be copied safely, but no page map
@@ -413,6 +412,120 @@ backup_data_file(const char *from_root, const char *to_root,
 	return true;
 }
 
+/*
+ * Restore compressed file that was backed up partly.
+ * 
+ */
+static void
+restore_file_partly(const char *from_root,const char *to_root, pgFile *file)
+{
+	FILE	   *in;
+	FILE	   *out;
+	size_t		read_len = 0;
+	int			errno_tmp;
+	struct stat	st;
+	char		to_path[MAXPGPATH];
+	char		buf[8192];
+	size_t write_size = 0;
+
+	join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
+	/* open backup mode file for read */
+	in = fopen(file->path, "r");
+	if (in == NULL)
+	{
+		elog(ERROR, "cannot open backup file \"%s\": %s", file->path,
+			strerror(errno));
+	}
+	out = fopen(to_path, "r+");
+
+	/* stat source file to change mode of destination file */
+	if (fstat(fileno(in), &st) == -1)
+	{
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot stat \"%s\": %s", file->path,
+			 strerror(errno));
+	}
+
+	if (fseek(out, 0, SEEK_END) < 0)
+		elog(ERROR, "cannot seek END of \"%s\": %s",
+				to_path, strerror(errno));
+
+	/* copy everything from backup to the end of the file */
+	for (;;)
+	{
+		if ((read_len = fread(buf, 1, sizeof(buf), in)) != sizeof(buf))
+			break;
+
+		if (fwrite(buf, 1, read_len, out) != read_len)
+		{
+			errno_tmp = errno;
+			/* oops */
+			fclose(in);
+			fclose(out);
+			elog(ERROR, "cannot write to \"%s\": %s", to_path,
+				 strerror(errno_tmp));
+		}
+		write_size += read_len;
+	}
+
+	errno_tmp = errno;
+	if (!feof(in))
+	{
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot read backup mode file \"%s\": %s",
+			 file->path, strerror(errno_tmp));
+	}
+
+	/* copy odd part. */
+	if (read_len > 0)
+	{
+		if (fwrite(buf, 1, read_len, out) != read_len)
+		{
+			errno_tmp = errno;
+			/* oops */
+			fclose(in);
+			fclose(out);
+			elog(ERROR, "cannot write to \"%s\": %s", to_path,
+				 strerror(errno_tmp));
+		}
+
+		write_size += read_len;
+	}
+
+// 	elog(LOG, "restore_file_partly(). %s write_size %lu, file->write_size %lu",
+// 			   file->path, write_size, file->write_size);
+
+	/* update file permission */
+	if (chmod(to_path, file->mode) == -1)
+	{
+		int errno_tmp = errno;
+
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot change mode of \"%s\": %s", to_path,
+			strerror(errno_tmp));
+	}
+
+	fclose(in);
+	fclose(out);
+}
+
+static void
+restore_compressed_file(const char *from_root,
+						const char *to_root,
+						pgFile *file)
+{
+	if (file->is_partial_copy == 0)
+		copy_file(from_root, to_root, file);
+	else if (file->is_partial_copy == 1)
+		restore_file_partly(from_root, to_root, file);
+	else
+		elog(ERROR, "restore_compressed_file(). Unknown is_partial_copy value %d",
+					file->is_partial_copy);
+}
+
 /*
  * Restore files in the from_root directory to the to_root directory with
  * same relative path.
@@ -429,10 +542,17 @@ restore_data_file(const char *from_root,
 	BackupPageHeader	header;
 	BlockNumber			blknum;
 
-	/* If the file is not a datafile, just copy it. */
 	if (!file->is_datafile)
 	{
+		/*
+		 * If the file is not a datafile and not compressed file,
+		 * just copy it.
+		 */
+		if (file->generation == -1)
 			copy_file(from_root, to_root, file);
+		else
+			restore_compressed_file(from_root, to_root, file);
+
 		return;
 	}
 
@@ -552,6 +672,16 @@ restore_data_file(const char *from_root,
 	fclose(out);
 }
 
+/* If someone's want to use this function before correct
+ * generation values is set, he can look up for corresponding
+ * .cfm file in the file_list
+ */
+bool
+is_compressed_data_file(pgFile *file)
+{
+	return (file->generation != -1);
+}
+
 bool
 copy_file(const char *from_root, const char *to_root, pgFile *file)
 {
@@ -681,6 +811,140 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
 	return true;
 }
 
+/*
+ * Save part of the file into backup.
+ * skip_size - size of the file in previous backup. We can skip it
+ *			   and copy just remaining part of the file
+ */
+bool
+copy_file_partly(const char *from_root, const char *to_root,
+				 pgFile *file, size_t skip_size)
+{
+	char		to_path[MAXPGPATH];
+	FILE	   *in;
+	FILE	   *out;
+	size_t		read_len = 0;
+	int			errno_tmp;
+	struct stat	st;
+	char		buf[8192];
+
+	/* reset size summary */
+	file->read_size = 0;
+	file->write_size = 0;
+
+	/* open backup mode file for read */
+	in = fopen(file->path, "r");
+	if (in == NULL)
+	{
+		/* maybe deleted, it's not error */
+		if (errno == ENOENT)
+			return false;
+
+		elog(ERROR, "cannot open source file \"%s\": %s", file->path,
+			 strerror(errno));
+	}
+
+	/* open backup file for write  */
+	if (check)
+		snprintf(to_path, lengthof(to_path), "%s/tmp", backup_path);
+	else
+		join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
+
+	out = fopen(to_path, "w");
+	if (out == NULL)
+	{
+		int errno_tmp = errno;
+		fclose(in);
+		elog(ERROR, "cannot open destination file \"%s\": %s",
+			 to_path, strerror(errno_tmp));
+	}
+
+	/* stat source file to change mode of destination file */
+	if (fstat(fileno(in), &st) == -1)
+	{
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot stat \"%s\": %s", file->path,
+			 strerror(errno));
+	}
+
+	if (fseek(in, skip_size, SEEK_SET) < 0)
+		elog(ERROR, "cannot seek %lu of \"%s\": %s",
+				skip_size, file->path, strerror(errno));
+
+	/*
+	 * copy content
+	 * NOTE: Now CRC is not computed for compressed files now.
+	 */
+	for (;;)
+	{
+		if ((read_len = fread(buf, 1, sizeof(buf), in)) != sizeof(buf))
+			break;
+
+		if (fwrite(buf, 1, read_len, out) != read_len)
+		{
+			errno_tmp = errno;
+			/* oops */
+			fclose(in);
+			fclose(out);
+			elog(ERROR, "cannot write to \"%s\": %s", to_path,
+				 strerror(errno_tmp));
+		}
+
+		file->write_size += sizeof(buf);
+		file->read_size += sizeof(buf);
+	}
+
+	errno_tmp = errno;
+	if (!feof(in))
+	{
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot read backup mode file \"%s\": %s",
+			 file->path, strerror(errno_tmp));
+	}
+
+	/* copy odd part. */
+	if (read_len > 0)
+	{
+		if (fwrite(buf, 1, read_len, out) != read_len)
+		{
+			errno_tmp = errno;
+			/* oops */
+			fclose(in);
+			fclose(out);
+			elog(ERROR, "cannot write to \"%s\": %s", to_path,
+				 strerror(errno_tmp));
+		}
+
+		file->write_size += read_len;
+		file->read_size += read_len;
+	}
+
+	/* update file permission */
+	if (chmod(to_path, st.st_mode) == -1)
+	{
+		errno_tmp = errno;
+		fclose(in);
+		fclose(out);
+		elog(ERROR, "cannot change mode of \"%s\": %s", to_path,
+			 strerror(errno_tmp));
+	}
+
+	/* add meta information needed for recovery */
+	file->is_partial_copy = 1;
+
+//	elog(LOG, "copy_file_partly(). %s file->write_size %lu", to_path, file->write_size);
+
+	fclose(in);
+	fclose(out);
+
+	if (check)
+		remove(to_path);
+
+	return true;
+}
+
 bool
 calc_file(pgFile *file)
 {

dir.c

@@ -92,6 +92,8 @@ pgFileNew(const char *path, bool omit_symlink)
 	file->ptrack_path = NULL;
 	file->segno = 0;
 	file->path = pgut_malloc(strlen(path) + 1);
+	file->generation = -1;
+	file->is_partial_copy = 0;
 	strcpy(file->path, path);		/* enough buffer size guaranteed */
 
 	return file;
@@ -514,13 +516,15 @@ dir_print_file_list(FILE *out, const parray *files, const char *root, const char
 				file->crc, file->mode & (S_IRWXU | S_IRWXG | S_IRWXO));
 
 		if (S_ISLNK(file->mode))
-			fprintf(out, " %s\n", file->linked);
+			fprintf(out, " %s", file->linked);
 		else
 		{
 			char timestamp[20];
 			time2iso(timestamp, 20, file->mtime);
-			fprintf(out, " %s\n", timestamp);
+			fprintf(out, " %s", timestamp);
 		}
+
+		fprintf(out, " %d %d\n", file->generation, file->is_partial_copy);
 	}
 }
 
@@ -546,6 +550,8 @@ dir_read_file_list(const char *root, const char *file_txt)
 	{
 		char			path[MAXPGPATH];
 		char			type;
+		int				generation = -1;
+		int				is_partial_copy = 0;
 		unsigned long	write_size;
 		pg_crc32		crc;
 		unsigned int	mode;	/* bit length of mode_t depends on platforms */
@@ -553,6 +559,13 @@ dir_read_file_list(const char *root, const char *file_txt)
 		pgFile			*file;
 
 		memset(&tm, 0, sizeof(tm));
+		if (sscanf(buf, "%s %c %lu %u %o %d-%d-%d %d:%d:%d %d %d",
+			path, &type, &write_size, &crc, &mode,
+			&tm.tm_year, &tm.tm_mon, &tm.tm_mday,
+			&tm.tm_hour, &tm.tm_min, &tm.tm_sec,
+			&generation, &is_partial_copy) != 13)
+		{
+			/* Maybe the file_list we're trying to recovery is in old format */
 			if (sscanf(buf, "%s %c %lu %u %o %d-%d-%d %d:%d:%d",
 				path, &type, &write_size, &crc, &mode,
 				&tm.tm_year, &tm.tm_mon, &tm.tm_mday,
@@ -561,6 +574,8 @@ dir_read_file_list(const char *root, const char *file_txt)
 				elog(ERROR, "invalid format found in \"%s\"",
 					file_txt);
 			}
+		}
+
 		if (type != 'f' && type != 'F' && type != 'd' && type != 'l')
 		{
 			elog(ERROR, "invalid type '%c' found in \"%s\"",
@@ -581,6 +596,8 @@ dir_read_file_list(const char *root, const char *file_txt)
 		file->mode = mode |
 			((type == 'f' || type == 'F') ? S_IFREG :
 			 type == 'd' ? S_IFDIR : type == 'l' ? S_IFLNK : 0);
+		file->generation = generation;
+		file->is_partial_copy = is_partial_copy;
 		file->size = 0;
 		file->read_size = 0;
 		file->write_size = write_size;

pg_probackup.h

@@ -25,6 +25,10 @@
 #include "storage/block.h"
 #include "storage/checksum.h"
 
+#ifndef WIN32
+#include <sys/mman.h>
+#endif
+
 /* Query to fetch current transaction ID */
 #define TXID_CURRENT_SQL	"SELECT txid_current();"
 #define TXID_CURRENT_IF_SQL	"SELECT txid_snapshot_xmax(txid_current_snapshot());"
@@ -66,6 +70,10 @@ typedef struct pgFile
 	char	*path;			/* path of the file */
 	char	*ptrack_path;
 	int		segno;			/* Segment number for ptrack */
+	int		generation;		/* Generation of compressed file.
+							 * -1 for non-compressed files */
+	int	is_partial_copy; /* for compressed files.
+						  * 1 if backed up via copy_file_partly()  */
 	volatile uint32 lock;
 	datapagemap_t pagemap;
 } pgFile;
@@ -166,6 +174,26 @@ typedef union DataPage
 	char			data[BLCKSZ];
 } DataPage;
 
+
+/*
+ * This struct definition mirrors one from cfs.h,
+ * but doesn't use atomic variables, since they are not allowed in
+ * frontend code.
+ */
+typedef struct
+{
+	uint32 physSize;
+	uint32 virtSize;
+	uint32 usedSize;
+	uint32 lock;
+	pid_t	postmasterPid;
+	uint64	generation;
+	uint64	inodes[RELSEG_SIZE];
+} FileMap;
+
+extern FileMap* cfs_mmap(int md);
+extern int cfs_munmap(FileMap* map);
+
 /*
  * return pointer that exceeds the length of prefix from character string.
  * ex. str="/xxx/yyy/zzz", prefix="/xxx/yyy", return="zzz".
@@ -309,8 +337,14 @@ extern bool backup_data_file(const char *from_root, const char *to_root,
 							 pgFile *file, const XLogRecPtr *lsn);
 extern void restore_data_file(const char *from_root, const char *to_root,
 							  pgFile *file, pgBackup *backup);
+extern bool is_compressed_data_file(pgFile *file);
+extern bool backup_compressed_file_partially(pgFile *file,
+											 void *arg,
+											 size_t *skip_size);
 extern bool copy_file(const char *from_root, const char *to_root,
 					  pgFile *file);
+extern bool copy_file_partly(const char *from_root, const char *to_root,
+				 pgFile *file, size_t skip_size);
 
 extern bool calc_file(pgFile *file);
 

validate.c

@@ -325,7 +325,11 @@ pgBackupValidateFiles(void *arg)
 			elog(ERROR, "interrupted during validate");
 
 		/* skipped backup while differential backup */
-		if (file->write_size == BYTES_INVALID || !S_ISREG(file->mode))
+		/* NOTE We don't compute checksums for compressed data,
+		 * so skip it too */
+		if (file->write_size == BYTES_INVALID
+			|| !S_ISREG(file->mode)
+			|| file->generation != -1)
 			continue;
 
 		/* print progress */