From 472c257d571b868196b6cf994df2c4e65ce1b79d Mon Sep 17 00:00:00 2001
From: Nitin Motiani <nitinmotiani@google.com>
Date: Tue, 11 Feb 2025 08:31:02 +0000
Subject: [PATCH v19 1/3] Add pipe support for directory mode of pg_dump and
 pg_restore

* We add a new flag --pipe which can be used in directory
  mode. This allows us to support multiple streams and we can
  do post processing like compression, filtering etc. This is
  currently not possible with directory-archive format.
* Currently this flag is only supported with compression none
  and archive format directory.
* This flag can't be used with the flag --file. Only one of the
  two flags can be used at a time.
* We reuse the filename field for the --pipe and also. And add a
  bool to specify that the field will be used as a pipe command.
* Most of the code remains as it is. The core change is that
  in case of --pipe, instead of fopen we do popen.
* The user would need a way to store the post-processing output
  in files. For that we support the same format as the directory
  mode currently does with the flag --file. We allow the user
  to add a format specifier %f to the --pipe-command. And for each
  stream, the format specifier is replaced with the corresponding
  file name. This file name is the same as it would have been if
  the flag --file had been used.
* To enable the above, there are a few places in the code where
  we change the file name creation logic. Currently the file name
  is appended to the directory name which is provided with --file flag.
  In case of --pipe, we instead replace %f with the file name.
  This change is made for the common use case and separately for
  blob files.
* The same change is made to pg_restore. It can be used
  to read from multiple streams and do pre-processing (decompression
  with a custom algorithm, filtering etc) before restore.
* Like pg_dump, here also either filename or --pipe
  set. If neither are set, the standard input is used as before.
* The --pipe command works -l and -L option in pg_restore.
* We also change the mode for large objects TOC from 'ab' to 'w'. 'ab'
  doesn't work with popen and based on the current code, it's not
  really required.
---
 src/bin/pg_dump/compress_gzip.c       |  9 ++-
 src/bin/pg_dump/compress_gzip.h       |  3 +-
 src/bin/pg_dump/compress_io.c         | 55 ++++++++++------
 src/bin/pg_dump/compress_io.h         | 16 ++++-
 src/bin/pg_dump/compress_lz4.c        | 11 +++-
 src/bin/pg_dump/compress_lz4.h        |  3 +-
 src/bin/pg_dump/compress_none.c       | 77 ++++++++++++++++++++--
 src/bin/pg_dump/compress_none.h       |  3 +-
 src/bin/pg_dump/compress_zstd.c       | 10 ++-
 src/bin/pg_dump/compress_zstd.h       |  3 +-
 src/bin/pg_dump/pg_backup.h           |  5 +-
 src/bin/pg_dump/pg_backup_archiver.c  | 46 +++++++++++---
 src/bin/pg_dump/pg_backup_archiver.h  |  2 +
 src/bin/pg_dump/pg_backup_directory.c | 92 +++++++++++++++++++++------
 src/bin/pg_dump/pg_dump.c             | 33 ++++++++--
 src/bin/pg_dump/pg_dumpall.c          |  8 +++
 src/bin/pg_dump/pg_restore.c          | 42 ++++++++++--
 17 files changed, 341 insertions(+), 77 deletions(-)

diff --git a/src/bin/pg_dump/compress_gzip.c b/src/bin/pg_dump/compress_gzip.c
index bd1043b2366..26bc137196c 100644
--- a/src/bin/pg_dump/compress_gzip.c
+++ b/src/bin/pg_dump/compress_gzip.c
@@ -429,8 +429,12 @@ Gzip_open_write(const char *path, const char *mode, CompressFileHandle *CFH)
 
 void
 InitCompressFileHandleGzip(CompressFileHandle *CFH,
-						   const pg_compress_specification compression_spec)
+						   const pg_compress_specification compression_spec,
+						   bool is_pipe)
 {
+	if (is_pipe)
+		pg_fatal("cPipe command not supported for Gzip");
+
 	CFH->open_func = Gzip_open;
 	CFH->open_write_func = Gzip_open_write;
 	CFH->read_func = Gzip_read;
@@ -455,7 +459,8 @@ InitCompressorGzip(CompressorState *cs,
 
 void
 InitCompressFileHandleGzip(CompressFileHandle *CFH,
-						   const pg_compress_specification compression_spec)
+						   const pg_compress_specification compression_spec,
+						   bool is_pipe)
 {
 	pg_fatal("this build does not support compression with %s", "gzip");
 }
diff --git a/src/bin/pg_dump/compress_gzip.h b/src/bin/pg_dump/compress_gzip.h
index af1a2a3445e..952c9223836 100644
--- a/src/bin/pg_dump/compress_gzip.h
+++ b/src/bin/pg_dump/compress_gzip.h
@@ -19,6 +19,7 @@
 extern void InitCompressorGzip(CompressorState *cs,
 							   const pg_compress_specification compression_spec);
 extern void InitCompressFileHandleGzip(CompressFileHandle *CFH,
-									   const pg_compress_specification compression_spec);
+									   const pg_compress_specification compression_spec,
+									   bool is_pipe);
 
 #endif							/* _COMPRESS_GZIP_H_ */
diff --git a/src/bin/pg_dump/compress_io.c b/src/bin/pg_dump/compress_io.c
index 52652b0d979..b4d84ef17d1 100644
--- a/src/bin/pg_dump/compress_io.c
+++ b/src/bin/pg_dump/compress_io.c
@@ -191,20 +191,28 @@ free_keep_errno(void *p)
  * Initialize a compress file handle for the specified compression algorithm.
  */
 CompressFileHandle *
-InitCompressFileHandle(const pg_compress_specification compression_spec)
+InitCompressFileHandle(const pg_compress_specification compression_spec,
+					   bool is_pipe)
 {
 	CompressFileHandle *CFH;
 
 	CFH = pg_malloc0_object(CompressFileHandle);
 
-	if (compression_spec.algorithm == PG_COMPRESSION_NONE)
-		InitCompressFileHandleNone(CFH, compression_spec);
+	/*
+	 * Always set to non-compressed when is_pipe assuming that external
+	 * compressor as part of pipe is more efficient. Can review in the future.
+	 */
+	if (is_pipe)
+		InitCompressFileHandleNone(CFH, compression_spec, is_pipe);
+
+	else if (compression_spec.algorithm == PG_COMPRESSION_NONE)
+		InitCompressFileHandleNone(CFH, compression_spec, is_pipe);
 	else if (compression_spec.algorithm == PG_COMPRESSION_GZIP)
-		InitCompressFileHandleGzip(CFH, compression_spec);
+		InitCompressFileHandleGzip(CFH, compression_spec, is_pipe);
 	else if (compression_spec.algorithm == PG_COMPRESSION_LZ4)
-		InitCompressFileHandleLZ4(CFH, compression_spec);
+		InitCompressFileHandleLZ4(CFH, compression_spec, is_pipe);
 	else if (compression_spec.algorithm == PG_COMPRESSION_ZSTD)
-		InitCompressFileHandleZstd(CFH, compression_spec);
+		InitCompressFileHandleZstd(CFH, compression_spec, is_pipe);
 
 	return CFH;
 }
@@ -237,7 +245,8 @@ check_compressed_file(const char *path, char **fname, char *ext)
  * On failure, return NULL with an error code in errno.
  */
 CompressFileHandle *
-InitDiscoverCompressFileHandle(const char *path, const char *mode)
+InitDiscoverCompressFileHandle(const char *path, const char *mode,
+							   bool is_pipe)
 {
 	CompressFileHandle *CFH = NULL;
 	struct stat st;
@@ -250,25 +259,31 @@ InitDiscoverCompressFileHandle(const char *path, const char *mode)
 
 	fname = pg_strdup(path);
 
-	if (hasSuffix(fname, ".gz"))
-		compression_spec.algorithm = PG_COMPRESSION_GZIP;
-	else if (hasSuffix(fname, ".lz4"))
-		compression_spec.algorithm = PG_COMPRESSION_LZ4;
-	else if (hasSuffix(fname, ".zst"))
-		compression_spec.algorithm = PG_COMPRESSION_ZSTD;
-	else
+	/*
+	 * If the path is a pipe command, the compression algorithm is none.
+	 */
+	if (!is_pipe)
 	{
-		if (stat(path, &st) == 0)
-			compression_spec.algorithm = PG_COMPRESSION_NONE;
-		else if (check_compressed_file(path, &fname, "gz"))
+		if (hasSuffix(fname, ".gz"))
 			compression_spec.algorithm = PG_COMPRESSION_GZIP;
-		else if (check_compressed_file(path, &fname, "lz4"))
+		else if (hasSuffix(fname, ".lz4"))
 			compression_spec.algorithm = PG_COMPRESSION_LZ4;
-		else if (check_compressed_file(path, &fname, "zst"))
+		else if (hasSuffix(fname, ".zst"))
 			compression_spec.algorithm = PG_COMPRESSION_ZSTD;
+		else
+		{
+			if (stat(path, &st) == 0)
+				compression_spec.algorithm = PG_COMPRESSION_NONE;
+			else if (check_compressed_file(path, &fname, "gz"))
+				compression_spec.algorithm = PG_COMPRESSION_GZIP;
+			else if (check_compressed_file(path, &fname, "lz4"))
+				compression_spec.algorithm = PG_COMPRESSION_LZ4;
+			else if (check_compressed_file(path, &fname, "zst"))
+				compression_spec.algorithm = PG_COMPRESSION_ZSTD;
+		}
 	}
 
-	CFH = InitCompressFileHandle(compression_spec);
+	CFH = InitCompressFileHandle(compression_spec, is_pipe);
 	errno = 0;
 	if (!CFH->open_func(fname, -1, mode, CFH))
 	{
diff --git a/src/bin/pg_dump/compress_io.h b/src/bin/pg_dump/compress_io.h
index ed7b14f0963..463e3047654 100644
--- a/src/bin/pg_dump/compress_io.h
+++ b/src/bin/pg_dump/compress_io.h
@@ -186,6 +186,16 @@ struct CompressFileHandle
 	 */
 	pg_compress_specification compression_spec;
 
+	/*
+	 * Compression specification for this file handle.
+	 */
+	bool		is_pipe;
+
+	/*
+	 * The command path or template (only used if is_pipe is true).
+	 */
+	char	   *command;
+
 	/*
 	 * Private data to be used by the compressor.
 	 */
@@ -195,7 +205,8 @@ struct CompressFileHandle
 /*
  * Initialize a compress file handle with the requested compression.
  */
-extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec);
+extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specification compression_spec,
+												  bool is_pipe);
 
 /*
  * Initialize a compress file stream. Infer the compression algorithm
@@ -203,6 +214,7 @@ extern CompressFileHandle *InitCompressFileHandle(const pg_compress_specificatio
  * suffixes in 'path'.
  */
 extern CompressFileHandle *InitDiscoverCompressFileHandle(const char *path,
-														  const char *mode);
+														  const char *mode,
+														  bool is_pipe);
 extern bool EndCompressFileHandle(CompressFileHandle *CFH);
 #endif
diff --git a/src/bin/pg_dump/compress_lz4.c b/src/bin/pg_dump/compress_lz4.c
index 500d5e16a6d..6100d04cd81 100644
--- a/src/bin/pg_dump/compress_lz4.c
+++ b/src/bin/pg_dump/compress_lz4.c
@@ -766,10 +766,14 @@ LZ4Stream_open_write(const char *path, const char *mode, CompressFileHandle *CFH
  */
 void
 InitCompressFileHandleLZ4(CompressFileHandle *CFH,
-						  const pg_compress_specification compression_spec)
+						  const pg_compress_specification compression_spec,
+						  bool is_pipe)
 {
 	LZ4State   *state;
 
+	if (is_pipe)
+		pg_fatal("Pipe command not supported for LZ4");
+
 	CFH->open_func = LZ4Stream_open;
 	CFH->open_write_func = LZ4Stream_open_write;
 	CFH->read_func = LZ4Stream_read;
@@ -785,6 +789,8 @@ InitCompressFileHandleLZ4(CompressFileHandle *CFH,
 	if (CFH->compression_spec.level >= 0)
 		state->prefs.compressionLevel = CFH->compression_spec.level;
 
+	CFH->is_pipe = is_pipe;
+
 	CFH->private_data = state;
 }
 #else							/* USE_LZ4 */
@@ -797,7 +803,8 @@ InitCompressorLZ4(CompressorState *cs,
 
 void
 InitCompressFileHandleLZ4(CompressFileHandle *CFH,
-						  const pg_compress_specification compression_spec)
+						  const pg_compress_specification compression_spec,
+						  bool is_pipe)
 {
 	pg_fatal("this build does not support compression with %s", "LZ4");
 }
diff --git a/src/bin/pg_dump/compress_lz4.h b/src/bin/pg_dump/compress_lz4.h
index 7360a469fc0..2c235cf3a50 100644
--- a/src/bin/pg_dump/compress_lz4.h
+++ b/src/bin/pg_dump/compress_lz4.h
@@ -19,6 +19,7 @@
 extern void InitCompressorLZ4(CompressorState *cs,
 							  const pg_compress_specification compression_spec);
 extern void InitCompressFileHandleLZ4(CompressFileHandle *CFH,
-									  const pg_compress_specification compression_spec);
+									  const pg_compress_specification compression_spec,
+									  bool is_pipe);
 
 #endif							/* _COMPRESS_LZ4_H_ */
diff --git a/src/bin/pg_dump/compress_none.c b/src/bin/pg_dump/compress_none.c
index 47bc6f1f3dd..02bf9a507ba 100644
--- a/src/bin/pg_dump/compress_none.c
+++ b/src/bin/pg_dump/compress_none.c
@@ -14,6 +14,7 @@
 #include "postgres_fe.h"
 #include <unistd.h>
 
+#include "port.h"
 #include "compress_none.h"
 #include "pg_backup_utils.h"
 
@@ -210,10 +211,40 @@ close_none(CompressFileHandle *CFH)
 
 	if (fp)
 	{
-		errno = 0;
-		ret = fclose(fp);
-		if (ret != 0)
-			pg_log_error("could not close file: %m");
+		if (CFH->is_pipe)
+		{
+			ret = pclose(fp);
+			if (ret != 0)
+			{
+				/*
+				 * For pipe commands, pclose() returns the exit status of the
+				 * child process. If the shell command itself fails (e.g.
+				 * "command not found"), pclose() will return a non-zero exit
+				 * status, but errno will likely remain 0 (Success). We use
+				 * wait_result_to_str to decode the status and pg_fatal to
+				 * prevent the caller from logging a generic and misleading
+				 * "could not close file: Success" message. We also include
+				 * the failing command string to help the user debug.
+				 */
+				char	   *reason = wait_result_to_str(ret);
+
+				pg_fatal("pipe command failed: \"%s\": %s",
+						 CFH->command ? CFH->command : "unknown",
+						 reason);
+			}
+		}
+		else
+		{
+			ret = fclose(fp);
+			if (ret != 0)
+				pg_fatal("could not close file: %m");
+		}
+	}
+
+	if (CFH->command)
+	{
+		pg_free(CFH->command);
+		CFH->command = NULL;
 	}
 
 	return ret == 0;
@@ -225,6 +256,23 @@ eof_none(CompressFileHandle *CFH)
 	return feof((FILE *) CFH->private_data) != 0;
 }
 
+static FILE *
+open_handle_none(const char *path, const char *mode, bool is_pipe)
+{
+	if (is_pipe)
+	{
+		/*
+		 * If the path is a pipe, we use popen(). Note that we do not track
+		 * the child PID for cleanup during fatal errors. We intentionally
+		 * rely on standard POSIX semantics: if pg_dump crashes, the OS will
+		 * close our end of the pipe, sending EOF to the child process, which
+		 * will then cleanly exit on its own.
+		 */
+		return popen(path, mode);
+	}
+	return fopen(path, mode);
+}
+
 static bool
 open_none(const char *path, int fd, const char *mode, CompressFileHandle *CFH)
 {
@@ -245,9 +293,13 @@ open_none(const char *path, int fd, const char *mode, CompressFileHandle *CFH)
 	}
 	else
 	{
-		CFH->private_data = fopen(path, mode);
+		CFH->private_data = open_handle_none(path, mode, CFH->is_pipe);
+
 		if (CFH->private_data == NULL)
 			return false;
+
+		if (CFH->is_pipe)
+			CFH->command = pg_strdup(path);
 	}
 
 	return true;
@@ -258,10 +310,17 @@ open_write_none(const char *path, const char *mode, CompressFileHandle *CFH)
 {
 	Assert(CFH->private_data == NULL);
 
-	CFH->private_data = fopen(path, mode);
+	pg_log_debug("Opening %s, pipe is %s",
+				 path, CFH->is_pipe ? "true" : "false");
+
+	CFH->private_data = open_handle_none(path, mode, CFH->is_pipe);
+
 	if (CFH->private_data == NULL)
 		return false;
 
+	if (CFH->is_pipe)
+		CFH->command = pg_strdup(path);
+
 	return true;
 }
 
@@ -271,7 +330,8 @@ open_write_none(const char *path, const char *mode, CompressFileHandle *CFH)
 
 void
 InitCompressFileHandleNone(CompressFileHandle *CFH,
-						   const pg_compress_specification compression_spec)
+						   const pg_compress_specification compression_spec,
+						   bool is_pipe)
 {
 	CFH->open_func = open_none;
 	CFH->open_write_func = open_write_none;
@@ -283,5 +343,8 @@ InitCompressFileHandleNone(CompressFileHandle *CFH,
 	CFH->eof_func = eof_none;
 	CFH->get_error_func = get_error_none;
 
+	CFH->is_pipe = is_pipe;
+
+	CFH->command = NULL;
 	CFH->private_data = NULL;
 }
diff --git a/src/bin/pg_dump/compress_none.h b/src/bin/pg_dump/compress_none.h
index 5134f012ee9..57943ceff7f 100644
--- a/src/bin/pg_dump/compress_none.h
+++ b/src/bin/pg_dump/compress_none.h
@@ -19,6 +19,7 @@
 extern void InitCompressorNone(CompressorState *cs,
 							   const pg_compress_specification compression_spec);
 extern void InitCompressFileHandleNone(CompressFileHandle *CFH,
-									   const pg_compress_specification compression_spec);
+									   const pg_compress_specification compression_spec,
+									   bool is_pipe);
 
 #endif							/* _COMPRESS_NONE_H_ */
diff --git a/src/bin/pg_dump/compress_zstd.c b/src/bin/pg_dump/compress_zstd.c
index 68f1d815917..57c4ad16500 100644
--- a/src/bin/pg_dump/compress_zstd.c
+++ b/src/bin/pg_dump/compress_zstd.c
@@ -27,7 +27,8 @@ InitCompressorZstd(CompressorState *cs, const pg_compress_specification compress
 }
 
 void
-InitCompressFileHandleZstd(CompressFileHandle *CFH, const pg_compress_specification compression_spec)
+InitCompressFileHandleZstd(CompressFileHandle *CFH, const pg_compress_specification compression_spec,
+						   bool is_pipe)
 {
 	pg_fatal("this build does not support compression with %s", "ZSTD");
 }
@@ -574,8 +575,12 @@ Zstd_get_error(CompressFileHandle *CFH)
 
 void
 InitCompressFileHandleZstd(CompressFileHandle *CFH,
-						   const pg_compress_specification compression_spec)
+						   const pg_compress_specification compression_spec,
+						   bool is_pipe)
 {
+	if (is_pipe)
+		pg_fatal("Pipe command not supported for Zstd");
+
 	CFH->open_func = Zstd_open;
 	CFH->open_write_func = Zstd_open_write;
 	CFH->read_func = Zstd_read;
@@ -587,6 +592,7 @@ InitCompressFileHandleZstd(CompressFileHandle *CFH,
 	CFH->get_error_func = Zstd_get_error;
 
 	CFH->compression_spec = compression_spec;
+	CFH->is_pipe = is_pipe;
 
 	CFH->private_data = NULL;
 }
diff --git a/src/bin/pg_dump/compress_zstd.h b/src/bin/pg_dump/compress_zstd.h
index 1222d7107d9..8b06657bc80 100644
--- a/src/bin/pg_dump/compress_zstd.h
+++ b/src/bin/pg_dump/compress_zstd.h
@@ -20,6 +20,7 @@
 extern void InitCompressorZstd(CompressorState *cs,
 							   const pg_compress_specification compression_spec);
 extern void InitCompressFileHandleZstd(CompressFileHandle *CFH,
-									   const pg_compress_specification compression_spec);
+									   const pg_compress_specification compression_spec,
+									   bool is_pipe);
 
 #endif							/* COMPRESS_ZSTD_H */
diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h
index c7bdda1deed..f13922335f9 100644
--- a/src/bin/pg_dump/pg_backup.h
+++ b/src/bin/pg_dump/pg_backup.h
@@ -316,14 +316,15 @@ extern void ProcessArchiveRestoreOptions(Archive *AHX);
 extern void RestoreArchive(Archive *AHX);
 
 /* Open an existing archive */
-extern Archive *OpenArchive(const char *FileSpec, const ArchiveFormat fmt);
+extern Archive *OpenArchive(const char *FileSpec, const ArchiveFormat fmt, bool is_pipe);
 
 /* Create a new archive */
 extern Archive *CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
 							  const pg_compress_specification compression_spec,
 							  bool dosync, ArchiveMode mode,
 							  SetupWorkerPtrType setupDumpWorker,
-							  DataDirSyncMethod sync_method);
+							  DataDirSyncMethod sync_method,
+							  bool is_pipe);
 
 /* The --list option */
 extern void PrintTOCSummary(Archive *AHX);
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index d7da3fc4325..1e08baff84d 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -57,7 +57,7 @@ static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt,
 							   const pg_compress_specification compression_spec,
 							   bool dosync, ArchiveMode mode,
 							   SetupWorkerPtrType setupWorkerPtr,
-							   DataDirSyncMethod sync_method);
+							   DataDirSyncMethod sync_method, bool FileSpecIsPipe);
 static void _getObjectDescription(PQExpBuffer buf, const TocEntry *te);
 static void _printTocEntry(ArchiveHandle *AH, TocEntry *te, const char *pfx);
 static void _doSetFixedOutputState(ArchiveHandle *AH);
@@ -233,11 +233,12 @@ CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
 			  const pg_compress_specification compression_spec,
 			  bool dosync, ArchiveMode mode,
 			  SetupWorkerPtrType setupDumpWorker,
-			  DataDirSyncMethod sync_method)
+			  DataDirSyncMethod sync_method,
+			  bool FileSpecIsPipe)
 
 {
 	ArchiveHandle *AH = _allocAH(FileSpec, fmt, compression_spec,
-								 dosync, mode, setupDumpWorker, sync_method);
+								 dosync, mode, setupDumpWorker, sync_method, FileSpecIsPipe);
 
 	return (Archive *) AH;
 }
@@ -245,7 +246,7 @@ CreateArchive(const char *FileSpec, const ArchiveFormat fmt,
 /* Open an existing archive */
 /* Public */
 Archive *
-OpenArchive(const char *FileSpec, const ArchiveFormat fmt)
+OpenArchive(const char *FileSpec, const ArchiveFormat fmt, bool FileSpecIsPipe)
 {
 	ArchiveHandle *AH;
 	pg_compress_specification compression_spec = {0};
@@ -253,7 +254,7 @@ OpenArchive(const char *FileSpec, const ArchiveFormat fmt)
 	compression_spec.algorithm = PG_COMPRESSION_NONE;
 	AH = _allocAH(FileSpec, fmt, compression_spec, true,
 				  archModeRead, setupRestoreWorker,
-				  DATA_DIR_SYNC_METHOD_FSYNC);
+				  DATA_DIR_SYNC_METHOD_FSYNC, FileSpecIsPipe);
 
 	return (Archive *) AH;
 }
@@ -1721,7 +1722,19 @@ SetOutput(ArchiveHandle *AH, const char *filename,
 	else
 		mode = PG_BINARY_W;
 
-	CFH = InitCompressFileHandle(compression_spec);
+	/*
+	 * The output handle (usually stdout) should never be a pipe command
+	 * managed by our popen logic, even if the archive itself is a pipe.  Our
+	 * pipe command implementation for directory mode is a template for the
+	 * data files, not for this primary output stream.
+	 *
+	 * Furthermore, marking this as a pipe command would cause it to be closed
+	 * with pclose() instead of fclose().  Since this handle is opened via
+	 * fdopen() (for stdout) or fopen() (for a regular file), using pclose()
+	 * on it is a bug that causes failures on BSD-based systems (like FreeBSD
+	 * or macOS).
+	 */
+	CFH = InitCompressFileHandle(compression_spec, false);
 
 	if (!CFH->open_func(filename, fn, mode, CFH))
 	{
@@ -2376,7 +2389,8 @@ static ArchiveHandle *
 _allocAH(const char *FileSpec, const ArchiveFormat fmt,
 		 const pg_compress_specification compression_spec,
 		 bool dosync, ArchiveMode mode,
-		 SetupWorkerPtrType setupWorkerPtr, DataDirSyncMethod sync_method)
+		 SetupWorkerPtrType setupWorkerPtr, DataDirSyncMethod sync_method,
+		 bool FileSpecIsPipe)
 {
 	ArchiveHandle *AH;
 	CompressFileHandle *CFH;
@@ -2417,6 +2431,8 @@ _allocAH(const char *FileSpec, const ArchiveFormat fmt,
 	else
 		AH->fSpec = NULL;
 
+	AH->is_pipe = FileSpecIsPipe;
+
 	AH->currUser = NULL;		/* unknown */
 	AH->currSchema = NULL;		/* ditto */
 	AH->currTablespace = NULL;	/* ditto */
@@ -2429,14 +2445,26 @@ _allocAH(const char *FileSpec, const ArchiveFormat fmt,
 
 	AH->mode = mode;
 	AH->compression_spec = compression_spec;
-	AH->dosync = dosync;
+	AH->dosync = FileSpecIsPipe ? false : dosync;
 	AH->sync_method = sync_method;
 
 	memset(&(AH->sqlparse), 0, sizeof(AH->sqlparse));
 
 	/* Open stdout with no compression for AH output handle */
 	out_compress_spec.algorithm = PG_COMPRESSION_NONE;
-	CFH = InitCompressFileHandle(out_compress_spec);
+
+	/*
+	 * The output handle (usually stdout) should never be a pipe command
+	 * managed by our popen logic, even if the archive itself is a pipe.  Our
+	 * pipe command implementation for directory mode is a template for the
+	 * data files, not for this primary output stream.
+	 *
+	 * Furthermore, marking this as a pipe command would cause it to be closed
+	 * with pclose() instead of fclose().  Since this handle is opened via
+	 * fdopen() (for stdout), using pclose() on it is a bug that causes
+	 * failures on BSD-based systems (like FreeBSD or macOS).
+	 */
+	CFH = InitCompressFileHandle(out_compress_spec, false);
 	if (!CFH->open_func(NULL, fileno(stdout), PG_BINARY_A, CFH))
 		pg_fatal("could not open stdout for appending: %m");
 	AH->OF = CFH;
diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h
index c1528d78853..e8fc4055e19 100644
--- a/src/bin/pg_dump/pg_backup_archiver.h
+++ b/src/bin/pg_dump/pg_backup_archiver.h
@@ -301,6 +301,8 @@ struct _archiveHandle
 	int			loCount;		/* # of LOs restored */
 
 	char	   *fSpec;			/* Archive File Spec */
+	bool		is_pipe;		/* fSpec is a pipe command template requiring
+								 * replacing %f with file name */
 	FILE	   *FH;				/* General purpose file handle */
 	void	   *OF;				/* Output file */
 
diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c
index c0b50223cec..fc939cfce6f 100644
--- a/src/bin/pg_dump/pg_backup_directory.c
+++ b/src/bin/pg_dump/pg_backup_directory.c
@@ -39,7 +39,8 @@
 #include <dirent.h>
 #include <sys/stat.h>
 
-#include "common/file_utils.h"
+/* #include "common/file_utils.h" */
+#include "common/percentrepl.h"
 #include "compress_io.h"
 #include "dumputils.h"
 #include "parallel.h"
@@ -157,8 +158,11 @@ InitArchiveFmt_Directory(ArchiveHandle *AH)
 
 	if (AH->mode == archModeWrite)
 	{
-		/* we accept an empty existing directory */
-		create_or_open_dir(ctx->directory);
+		if (!AH->is_pipe)		/* no checks for pipe */
+		{
+			/* we accept an empty existing directory */
+			create_or_open_dir(ctx->directory);
+		}
 	}
 	else
 	{							/* Read Mode */
@@ -167,7 +171,7 @@ InitArchiveFmt_Directory(ArchiveHandle *AH)
 
 		setFilePath(AH, fname, "toc.dat");
 
-		tocFH = InitDiscoverCompressFileHandle(fname, PG_BINARY_R);
+		tocFH = InitDiscoverCompressFileHandle(fname, PG_BINARY_R, AH->is_pipe);
 		if (tocFH == NULL)
 			pg_fatal("could not open input file \"%s\": %m", fname);
 
@@ -295,7 +299,7 @@ _StartData(ArchiveHandle *AH, TocEntry *te)
 
 	setFilePath(AH, fname, tctx->filename);
 
-	ctx->dataFH = InitCompressFileHandle(AH->compression_spec);
+	ctx->dataFH = InitCompressFileHandle(AH->compression_spec, AH->is_pipe);
 
 	if (!ctx->dataFH->open_write_func(fname, PG_BINARY_W, ctx->dataFH))
 		pg_fatal("could not open output file \"%s\": %m", fname);
@@ -353,7 +357,7 @@ _PrintFileData(ArchiveHandle *AH, char *filename)
 	if (!filename)
 		return;
 
-	CFH = InitDiscoverCompressFileHandle(filename, PG_BINARY_R);
+	CFH = InitDiscoverCompressFileHandle(filename, PG_BINARY_R, AH->is_pipe);
 	if (!CFH)
 		pg_fatal("could not open input file \"%s\": %m", filename);
 
@@ -416,7 +420,7 @@ _LoadLOs(ArchiveHandle *AH, TocEntry *te)
 	else
 		setFilePath(AH, tocfname, tctx->filename);
 
-	CFH = ctx->LOsTocFH = InitDiscoverCompressFileHandle(tocfname, PG_BINARY_R);
+	CFH = ctx->LOsTocFH = InitDiscoverCompressFileHandle(tocfname, PG_BINARY_R, AH->is_pipe);
 
 	if (ctx->LOsTocFH == NULL)
 		pg_fatal("could not open large object TOC file \"%s\" for input: %m",
@@ -434,7 +438,9 @@ _LoadLOs(ArchiveHandle *AH, TocEntry *te)
 					 tocfname, line);
 
 		StartRestoreLO(AH, oid, AH->public.ropt->dropSchema);
-		snprintf(path, MAXPGPATH, "%s/%s", ctx->directory, lofname);
+
+		setFilePath(AH, path, lofname);
+
 		_PrintFileData(AH, path);
 		EndRestoreLO(AH, oid);
 	}
@@ -545,7 +551,7 @@ _CloseArchive(ArchiveHandle *AH)
 
 		/* The TOC is always created uncompressed */
 		compression_spec.algorithm = PG_COMPRESSION_NONE;
-		tocFH = InitCompressFileHandle(compression_spec);
+		tocFH = InitCompressFileHandle(compression_spec, AH->is_pipe);
 		if (!tocFH->open_write_func(fname, PG_BINARY_W, tocFH))
 			pg_fatal("could not open output file \"%s\": %m", fname);
 		ctx->dataFH = tocFH;
@@ -606,13 +612,34 @@ _StartLOs(ArchiveHandle *AH, TocEntry *te)
 	lclTocEntry *tctx = (lclTocEntry *) te->formatData;
 	pg_compress_specification compression_spec = {0};
 	char		fname[MAXPGPATH];
+	const char *mode;
 
 	setFilePath(AH, fname, tctx->filename);
 
 	/* The LO TOC file is never compressed */
 	compression_spec.algorithm = PG_COMPRESSION_NONE;
-	ctx->LOsTocFH = InitCompressFileHandle(compression_spec);
-	if (!ctx->LOsTocFH->open_write_func(fname, "ab", ctx->LOsTocFH))
+	ctx->LOsTocFH = InitCompressFileHandle(compression_spec, AH->is_pipe);
+
+	/*
+	 * We use 'w' (PG_BINARY_W) mode for the LOs TOC file in all cases.
+	 * Historically, the mode for this file was "ab". However, append mode is
+	 * entirely redundant due to how large objects are partitioned.
+	 *
+	 * pg_dump splits large objects into chunks of up to 1000 blobs per
+	 * archive entry. Each chunk receives a completely unique dumpId, and the
+	 * TOC file is named using that ID (e.g., blobs_123.toc). Furthermore,
+	 * WriteDataChunksForTocEntry ensures a strict sequential lifecycle for
+	 * each entry: it calls _StartLOs (opens the file), then the dumper
+	 * function (writes the chunk), and finally _EndLOs (closes the file).
+	 *
+	 * Because a blobs_NNN.toc file is guaranteed to be unique and is only
+	 * opened exactly once, written to sequentially, and then closed forever,
+	 * there is no scenario where "ab" is required. This change to "w" is
+	 * necessary because popen() for pipe-commands only supports "r" and "w".
+	 */
+	mode = PG_BINARY_W;
+
+	if (!ctx->LOsTocFH->open_write_func(fname, mode, ctx->LOsTocFH))
 		pg_fatal("could not open output file \"%s\": %m", fname);
 }
 
@@ -626,10 +653,12 @@ _StartLO(ArchiveHandle *AH, TocEntry *te, Oid oid)
 {
 	lclContext *ctx = (lclContext *) AH->formatData;
 	char		fname[MAXPGPATH];
+	char		blob_name[MAXPGPATH];
 
-	snprintf(fname, MAXPGPATH, "%s/blob_%u.dat", ctx->directory, oid);
+	snprintf(blob_name, MAXPGPATH, "blob_%u.dat", oid);
+	setFilePath(AH, fname, blob_name);
 
-	ctx->dataFH = InitCompressFileHandle(AH->compression_spec);
+	ctx->dataFH = InitCompressFileHandle(AH->compression_spec, AH->is_pipe);
 	if (!ctx->dataFH->open_write_func(fname, PG_BINARY_W, ctx->dataFH))
 		pg_fatal("could not open output file \"%s\": %m", fname);
 }
@@ -683,15 +712,40 @@ setFilePath(ArchiveHandle *AH, char *buf, const char *relativeFilename)
 {
 	lclContext *ctx = (lclContext *) AH->formatData;
 	char	   *dname;
+	char	   *pipe;
 
 	dname = ctx->directory;
 
-	if (strlen(dname) + 1 + strlen(relativeFilename) + 1 > MAXPGPATH)
-		pg_fatal("file name too long: \"%s\"", dname);
 
-	strcpy(buf, dname);
-	strcat(buf, "/");
-	strcat(buf, relativeFilename);
+	if (AH->is_pipe)
+	{
+		/*
+		 * Unlike commands synthesized by the backend, this is a user-provided
+		 * template running client-side. We perform literal substitution
+		 * rather than using appendShellString() to avoid interfering with the
+		 * user's intentional shell quoting (e.g., for Windows vs Unix
+		 * differences). Since this is a client-side execution, there are no
+		 * privilege escalation concerns.
+		 */
+		pipe = replace_percent_placeholders(dname, "pipe", "f", relativeFilename);
+
+		if (strlen(pipe) >= MAXPGPATH)
+			pg_fatal("pipe command too long: \"%s\"", pipe);
+
+		strcpy(buf, pipe);
+
+		pfree(pipe);
+	}
+	else						/* replace all ocurrences of %f in dname with
+								 * relativeFilename */
+	{
+		if (strlen(dname) + 1 + strlen(relativeFilename) + 1 > MAXPGPATH)
+			pg_fatal("file name too long: \"%s\"", dname);
+
+		strcpy(buf, dname);
+		strcat(buf, "/");
+		strcat(buf, relativeFilename);
+	}
 }
 
 /*
@@ -743,7 +797,9 @@ _PrepParallelRestore(ArchiveHandle *AH)
 			else if (AH->compression_spec.algorithm == PG_COMPRESSION_LZ4)
 				strlcat(fname, ".lz4", sizeof(fname));
 			else if (AH->compression_spec.algorithm == PG_COMPRESSION_ZSTD)
+			{
 				strlcat(fname, ".zst", sizeof(fname));
+			}
 
 			if (stat(fname, &st) == 0)
 				te->dataLength = st.st_size;
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index feed88f9854..2b1131f3a4b 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -419,6 +419,8 @@ main(int argc, char **argv)
 {
 	int			c;
 	const char *filename = NULL;
+	char	   *pipe_command = NULL;
+	bool		is_pipe = false;
 	const char *format = "p";
 	TableInfo  *tblinfo;
 	int			numTables;
@@ -535,6 +537,7 @@ main(int argc, char **argv)
 		{"exclude-extension", required_argument, NULL, 17},
 		{"sequence-data", no_argument, &dopt.sequence_data, 1},
 		{"restrict-key", required_argument, NULL, 25},
+		{"pipe", required_argument, NULL, 26},
 
 		{NULL, 0, NULL, 0}
 	};
@@ -607,6 +610,8 @@ main(int argc, char **argv)
 
 			case 'f':
 				filename = pg_strdup(optarg);
+				is_pipe = false;	/* it already is, setting again here just
+									 * for clarity */
 				break;
 
 			case 'F':
@@ -799,6 +804,11 @@ main(int argc, char **argv)
 				dopt.restrict_key = pg_strdup(optarg);
 				break;
 
+			case 26:			/* pipe command */
+				pipe_command = pg_strdup(optarg);
+				is_pipe = true;
+				break;
+
 			default:
 				/* getopt_long already emitted a complaint */
 				pg_log_error_hint("Try \"%s --help\" for more information.", progname);
@@ -806,6 +816,10 @@ main(int argc, char **argv)
 		}
 	}
 
+	if (filename && pipe_command)
+		pg_fatal("options %s and %s cannot be used together",
+				 "-f/--file", "--pipe");
+
 	/*
 	 * Non-option argument specifies database name as long as it wasn't
 	 * already specified with -d / --dbname
@@ -907,14 +921,23 @@ main(int argc, char **argv)
 	else if (dopt.restrict_key)
 		pg_fatal("option %s can only be used with %s",
 				 "--restrict-key", "--format=plain");
+	if (is_pipe && archiveFormat != archDirectory)
+		pg_fatal("option --pipe is only supported with directory format");
+
+	if (is_pipe && strcmp(compression_algorithm_str, "none") != 0)
+		pg_fatal("option --pipe is not supported with any compression type");
+
+	if (is_pipe && numWorkers > 1 && strstr(pipe_command, "%f") == NULL)
+		pg_log_warning("parallel jobs with --pipe usually require the \"%%f\" placeholder to avoid data corruption from multiple workers writing to the same file");
 
 	/*
 	 * Custom and directory formats are compressed by default with gzip when
 	 * available, not the others.  If gzip is not available, no compression is
-	 * done by default.
+	 * done by default. If directory format is being used with pipe, no
+	 * compression is done.
 	 */
 	if ((archiveFormat == archCustom || archiveFormat == archDirectory) &&
-		!user_compression_defined)
+		!is_pipe && !user_compression_defined)
 	{
 #ifdef HAVE_LIBZ
 		compression_algorithm_str = "gzip";
@@ -963,8 +986,8 @@ main(int argc, char **argv)
 		pg_fatal("parallel backup only supported by the directory format");
 
 	/* Open the output file */
-	fout = CreateArchive(filename, archiveFormat, compression_spec,
-						 dosync, archiveMode, setupDumpWorker, sync_method);
+	fout = CreateArchive(is_pipe ? pipe_command : filename, archiveFormat, compression_spec,
+						 dosync, archiveMode, setupDumpWorker, sync_method, is_pipe);
 
 	/* Make dump options accessible right away */
 	SetArchiveOptions(fout, &dopt, NULL);
@@ -1296,6 +1319,8 @@ help(const char *progname)
 
 	printf(_("\nGeneral options:\n"));
 	printf(_("  -f, --file=FILENAME          output file or directory name\n"));
+	printf(_("  --pipe=COMMAND               execute command for each output file and\n"
+			 "                               write data to it via pipe\n"));
 	printf(_("  -F, --format=c|d|t|p         output file format (custom, directory, tar,\n"
 			 "                               plain text (default))\n"));
 	printf(_("  -j, --jobs=NUM               use this many parallel jobs to dump\n"));
diff --git a/src/bin/pg_dump/pg_dumpall.c b/src/bin/pg_dump/pg_dumpall.c
index c53e77c2878..f650736b98f 100644
--- a/src/bin/pg_dump/pg_dumpall.c
+++ b/src/bin/pg_dump/pg_dumpall.c
@@ -274,6 +274,14 @@ main(int argc, char *argv[])
 				appendShellString(pgdumpopts, filename);
 				break;
 
+				/*
+				 * Note: support for --pipe is currently skipped for
+				 * pg_dumpall due to the complexity of avoiding path
+				 * collisions between multiple databases and coordinating
+				 * nested directory structures. This could be considered as a
+				 * future enhancement.
+				 */
+
 			case 'g':
 				globals_only = true;
 				break;
diff --git a/src/bin/pg_dump/pg_restore.c b/src/bin/pg_dump/pg_restore.c
index 84b8d410c9e..f4836368aa0 100644
--- a/src/bin/pg_dump/pg_restore.c
+++ b/src/bin/pg_dump/pg_restore.c
@@ -1,5 +1,5 @@
 /*-------------------------------------------------------------------------
- *
+*
  * pg_restore.c
  *	pg_restore is an utility extracting postgres database definitions
  *	from a backup archive created by pg_dump using the archiver
@@ -63,9 +63,11 @@ main(int argc, char **argv)
 	int			exit_code;
 	int			numWorkers = 1;
 	Archive    *AH;
-	char	   *inputFileSpec;
+	char	   *inputFileSpec = NULL;
+	char	   *pipe_command = NULL;
 	bool		data_only = false;
 	bool		schema_only = false;
+	bool		is_pipe = false;
 	static int	disable_triggers = 0;
 	static int	enable_row_security = 0;
 	static int	if_exists = 0;
@@ -142,6 +144,7 @@ main(int argc, char **argv)
 		{"statistics-only", no_argument, &statistics_only, 1},
 		{"filter", required_argument, NULL, 4},
 		{"restrict-key", required_argument, NULL, 6},
+		{"pipe", required_argument, NULL, 7},
 
 		{NULL, 0, NULL, 0}
 	};
@@ -321,6 +324,11 @@ main(int argc, char **argv)
 				opts->restrict_key = pg_strdup(optarg);
 				break;
 
+			case 7:				/* pipe */
+				pipe_command = pg_strdup(optarg);
+				is_pipe = true;
+				break;
+
 			default:
 				/* getopt_long already emitted a complaint */
 				pg_log_error_hint("Try \"%s --help\" for more information.", progname);
@@ -328,11 +336,25 @@ main(int argc, char **argv)
 		}
 	}
 
-	/* Get file name from command line */
+	/*
+	 * Get file name from command line. Note that filename argument and pipe
+	 * can't both be set.
+	 */
 	if (optind < argc)
+	{
+		if (is_pipe)
+			pg_fatal("cannot specify both an input file and --pipe");
 		inputFileSpec = argv[optind++];
-	else
+	}
+
+	/*
+	 * Even if the file argument is not provided, if the pipe is specified, we
+	 * need to use that as the file arg and not fallback to stdio.
+	 */
+	else if (!is_pipe)
+	{
 		inputFileSpec = NULL;
+	}
 
 	/* Complain if any arguments remain */
 	if (optind < argc)
@@ -484,8 +506,16 @@ main(int argc, char **argv)
 			pg_fatal("unrecognized archive format \"%s\"; please specify \"c\", \"d\", or \"t\"",
 					 opts->formatName);
 	}
+	else
+		opts->format = archUnknown;
+
+	if (is_pipe && opts->format != archDirectory)
+		pg_fatal("option --pipe is only supported with directory format");
+
+	if (is_pipe && numWorkers > 1 && strstr(pipe_command, "%f") == NULL)
+		pg_log_warning("parallel jobs with --pipe usually require the \"%%f\" placeholder to avoid data corruption from multiple workers reading from the same file");
 
-	AH = OpenArchive(inputFileSpec, opts->format);
+	AH = OpenArchive(is_pipe ? pipe_command : inputFileSpec, opts->format, is_pipe);
 
 	SetArchiveOptions(AH, NULL, opts);
 
@@ -539,6 +569,8 @@ usage(const char *progname)
 	printf(_("\nGeneral options:\n"));
 	printf(_("  -d, --dbname=NAME        connect to database name\n"));
 	printf(_("  -f, --file=FILENAME      output file name (- for stdout)\n"));
+	printf(_("  --pipe=COMMAND           execute command for each input file and\n"
+			 "                           read data from it via pipe\n"));
 	printf(_("  -F, --format=c|d|t       backup file format (should be automatic)\n"));
 	printf(_("  -l, --list               print summarized TOC of the archive\n"));
 	printf(_("  -v, --verbose            verbose mode\n"));
-- 
2.55.0.699.gb54405d56f-goog

