[Issue #449] improve backward compatibility

Author: gsmolk

src/archive.c

@@ -105,6 +105,7 @@ static int push_file(WALSegno *xlogfile, const char *archive_status_dir,
 
 static parray *setup_push_filelist(const char *archive_status_dir,
 								   const char *first_file, int batch_size);
+static parray *setup_archive_subdirs(parray *batch_files, const char *archive_dir);
 
 static xlogFileType
 get_xlogFileType(const char *filename)
@@ -165,6 +166,7 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 
 	/* files to push in multi-thread mode */
 	parray     *batch_files = NULL;
+	parray     *archive_subdirs = NULL;
 	int         n_threads;
 
 	if (wal_file_name == NULL)
@@ -211,7 +213,19 @@ do_archive_push(InstanceState *instanceState, InstanceConfig *instance, char *wa
 						parray_num(batch_files), batch_size,
 						is_compress ? "zlib" : "none");
 
-	/* TODO: create subdirectories here, not in internal functions */
+	/* Extract subdirectories */
+	archive_subdirs = setup_archive_subdirs(batch_files, instanceState->instance_wal_subdir_path);
+	if (archive_subdirs)
+	{
+		for (i = 0; i < parray_num(archive_subdirs); i++)
+		{
+			char *subdir = (char *) parray_get(archive_subdirs, i);
+			if (fio_mkdir(subdir, DIR_PERMISSION, FIO_BACKUP_HOST) != 0)
+				elog(ERROR, "Cannot create subdirectory in WAL archive: '%s'", subdir);
+			pg_free(subdir);
+		}
+		parray_free(archive_subdirs);
+	}
 
 	num_threads = n_threads;
 
@@ -460,10 +474,6 @@ push_file_internal_uncompressed(const char *wal_file_name, const char *pg_xlog_d
 	/* calculate subdir in WAL archive */
 	get_archive_subdir(archive_subdir, archive_dir, wal_file_name, type);
 
-	/* create subdirectory */
-	if (fio_mkdir(archive_subdir, DIR_PERMISSION, FIO_BACKUP_HOST) != 0)
-		elog(ERROR, "Cannot create subdirectory in WAL archive: '%s'", archive_subdir);
-
 	/* to path */
 	join_path_components(to_fullpath, archive_subdir, wal_file_name);
 	canonicalize_path(to_fullpath);
@@ -711,10 +721,6 @@ push_file_internal_gz(const char *wal_file_name, const char *pg_xlog_dir,
 	/* calculate subdir in WAL archive */
 	get_archive_subdir(archive_subdir, archive_dir, wal_file_name, type);
 
-	/* create subdirectory */
-	if (fio_mkdir(archive_subdir, DIR_PERMISSION, FIO_BACKUP_HOST) != 0)
-		elog(ERROR, "Cannot create subdirectory in WAL archive: '%s'", archive_subdir);
-
 	/* to path */
 	join_path_components(to_fullpath, archive_subdir, wal_file_name);
 	canonicalize_path(to_fullpath);
@@ -1863,4 +1869,42 @@ get_archive_subdir(char *archive_subdir, const char *archive_dir, const char *wa
 
 	/* for all other files just use root directory of WAL archive */
 	strcpy(archive_subdir, archive_dir);
-}
+}
+
+/* Extract array of WAL archive subdirs using push filelist */
+parray*
+setup_archive_subdirs(parray *batch_files, const char *archive_dir)
+{
+	int     i;
+	parray *subdirs = NULL;
+	char   *cur_subdir = NULL;
+
+	/*
+	 * - Do we need to sort batch_files?
+	 * - No, we rely on sorting of status files
+	 */
+
+	for (i = 0; i < parray_num(batch_files); i++)
+	{
+		WALSegno *xlogfile = (WALSegno *) parray_get(batch_files, i);
+
+		if (xlogfile->type == SEGMENT || xlogfile->type == PARTIAL_SEGMENT || xlogfile->type == BACKUP_HISTORY_FILE)
+		{
+			char subdir[MAXPGPATH];
+
+			if (!subdirs)
+				subdirs = parray_new();
+
+			get_archive_subdir(subdir, archive_dir, xlogfile->name, xlogfile->type);
+
+			/* do not append the same subdir twice */
+			if (cur_subdir && strcmp(cur_subdir, subdir) == 0)
+				continue;
+
+			cur_subdir = pgut_strdup(subdir);
+			parray_append(subdirs, cur_subdir);
+		}
+	}
+
+	return subdirs;
+}

src/backup.c

@@ -1238,16 +1238,15 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
 {
 	XLogSegNo	targetSegNo;
 	char		wal_segment_path[MAXPGPATH],
-				wal_segment_subdir[MAXPGPATH], // used only to check file existence, not actual parsing
+//				wal_segment_subdir[MAXPGPATH], // used only to check file existence, not actual parsing
 				wal_segment[MAXFNAMELEN];
-	bool		file_exists = false;
 	uint32		try_count = 0,
 				timeout;
 	char		*wal_delivery_str = in_stream_dir ? "streamed":"archived";
 
-#ifdef HAVE_LIBZ
-	char		gz_wal_segment_path[MAXPGPATH];
-#endif
+//#ifdef HAVE_LIBZ
+//	char		gz_wal_segment_path[MAXPGPATH];
+//#endif
 
 	/* Compute the name of the WAL file containing requested LSN */
 	GetXLogSegNo(target_lsn, targetSegNo, instance_config.xlog_seg_size);
@@ -1257,12 +1256,12 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
 					instance_config.xlog_seg_size);
 
 	// obtain WAL archive subdir for ARCHIVE backup
-	if (in_stream_dir)
-		strcpy(wal_segment_subdir, wal_segment_dir);
-	else
-		get_archive_subdir(wal_segment_subdir, wal_segment_dir, wal_segment, SEGMENT);
-
-	join_path_components(wal_segment_path, wal_segment_subdir, wal_segment);
+//	if (in_stream_dir)
+//		strcpy(wal_segment_subdir, wal_segment_dir);
+//	else
+//		get_archive_subdir(wal_segment_subdir, wal_segment_dir, wal_segment, SEGMENT);
+//
+//	join_path_components(wal_segment_path, wal_segment_subdir, wal_segment);
 	/*
 	 * In pg_start_backup we wait for 'target_lsn' in 'pg_wal' directory if it is
 	 * stream and non-page backup. Page backup needs archived WAL files, so we
@@ -1283,30 +1282,15 @@ wait_wal_lsn(const char *wal_segment_dir, XLogRecPtr target_lsn, bool is_start_l
 		elog(LOG, "Looking for LSN %X/%X in segment: %s",
 			 (uint32) (target_lsn >> 32), (uint32) target_lsn, wal_segment);
 
-#ifdef HAVE_LIBZ
-	snprintf(gz_wal_segment_path, sizeof(gz_wal_segment_path), "%s.gz",
-			 wal_segment_path);
-#endif
+//#ifdef HAVE_LIBZ
+//	snprintf(gz_wal_segment_path, sizeof(gz_wal_segment_path), "%s.gz",
+//			 wal_segment_path);
+//#endif
 
 	/* Wait until target LSN is archived or streamed */
 	while (true)
 	{
-		if (!file_exists)
-		{
-			file_exists = fileExists(wal_segment_path, FIO_BACKUP_HOST);
-
-			/* Try to find compressed WAL file */
-			if (!file_exists)
-			{
-#ifdef HAVE_LIBZ
-				file_exists = fileExists(gz_wal_segment_path, FIO_BACKUP_HOST);
-				if (file_exists)
-					elog(LOG, "Found compressed WAL segment: %s", wal_segment_path);
-#endif
-			}
-			else
-				elog(LOG, "Found WAL segment: %s", wal_segment_path);
-		}
+		bool file_exists = IsWalFileExists(wal_segment, wal_segment_dir, in_stream_dir);
 
 		if (file_exists)
 		{

src/parsexlog.c

@@ -2047,4 +2047,64 @@ static XLogReaderState* WalReaderAllocate(uint32 wal_seg_size, XLogReaderData *r
 #else
 	return XLogReaderAllocate(&SimpleXLogPageRead, reader_data);
 #endif
-}
+}
+
+/*
+ * Is WAL file exists in archive directory
+ * first check subdirectory, then fallback to archive directory
+ */
+bool IsWalFileExists(const char *wal_segment_name, const char *wal_root_dir, bool in_stream_dir)
+{
+	char wal_file_fullpath[MAXPGPATH];
+	char wal_file_fullpath_gz[MAXPGPATH];
+	char wal_segment_subdir[MAXPGPATH];
+
+	if (in_stream_dir)
+	{
+		join_path_components(wal_file_fullpath, wal_root_dir, wal_segment_name);
+		if (fileExists(wal_file_fullpath, FIO_BACKUP_HOST))
+			goto found_uncompressed_file;
+
+		goto not_found;
+	}
+
+	/* obtain subdir in WAL archive */
+	get_archive_subdir(wal_segment_subdir, wal_root_dir, wal_segment_name, SEGMENT);
+
+	/* first try uncompressed segment in WAL archive subdir ... */
+	join_path_components(wal_file_fullpath, wal_segment_subdir, wal_segment_name);
+	if (fileExists(wal_file_fullpath, FIO_BACKUP_HOST))
+		goto found_uncompressed_file;
+
+#ifdef HAVE_LIBZ
+	/* ... fallback to compressed segment in WAL archive subdir ... */
+	snprintf(wal_file_fullpath_gz, MAXPGPATH, "%s.gz", wal_file_fullpath);
+	if (fileExists(wal_file_fullpath_gz, FIO_BACKUP_HOST))
+		goto found_compressed_file;
+#endif
+
+	/* ... fallback to uncompressed segment in archive dir ... */
+	join_path_components(wal_file_fullpath, wal_root_dir, wal_segment_name);
+	if (fileExists(wal_file_fullpath, FIO_BACKUP_HOST))
+		goto found_uncompressed_file;
+
+	/* ... fallback to compressed segment in archive dir */
+#ifdef HAVE_LIBZ
+	snprintf(wal_file_fullpath_gz, MAXPGPATH, "%s.gz", wal_file_fullpath);
+	if (fileExists(wal_file_fullpath_gz, FIO_BACKUP_HOST))
+		goto found_compressed_file;
+#endif
+
+	goto not_found;
+
+found_compressed_file:
+	elog(LOG, "Found WAL segment: %s", wal_file_fullpath);
+	return true;
+
+found_uncompressed_file:
+	elog(LOG, "Found compressed WAL segment: %s", wal_file_fullpath_gz);
+	return true;
+
+not_found:
+	return false;
+}

src/pg_probackup.h

@@ -649,7 +649,7 @@ typedef enum xlogFileType
 {
 	UNKNOWN,
 	SEGMENT,
-	TEMP_SEGMENT, // '.part' segment created by archive-push 
+	TEMP_SEGMENT, // '.part' segment created by archive-push
 	PARTIAL_SEGMENT,
 	HISTORY_FILE,
 	BACKUP_HISTORY_FILE
@@ -1053,6 +1053,7 @@ extern int dir_create_dir(const char *path, mode_t mode, bool strict);
 extern bool dir_is_empty(const char *path, fio_location location);
 
 extern bool fileExists(const char *path, fio_location location);
+extern bool IsWalFileExists(const char *wal_segment_name, const char *archive_dir, bool in_stream_dir);
 extern size_t pgFileSize(const char *path);
 
 extern pgFile *pgFileNew(const char *path, const char *rel_path,