pg_lake: streaming writes for INSERT, UPDATE, DELETE, COPY, and iceberg metadata

Adds an opt-in streaming-write feature that pushes bulk-write bytes
directly to pgduck_server via libpq COPY-IN, instead of writing them
to a shared filesystem under $PGDATA/pgsql_tmp and asking pgduck to
read them back. Companion to the pgduck_server RECEIVE protocol.

This decouples pgduck_server's filesystem from postgres's: the two
processes can run on different machines, different containers,
different pods, with no shared mount required.

User-visible:

- New GUC `pg_lake_engine.streaming_writes` (default off). When
  flipped on, the bulk-write code paths route through the streaming
  protocol; when off, all behavior is identical to today.
- No SQL syntax changes. INSERT, UPDATE, DELETE, COPY FROM STDIN,
  CREATE TABLE iceberg, and iceberg metadata uploads all switch
  paths transparently based on the GUC.

What this patch adds, by component:

`pg_lake_engine`:

- `OpenCSVStreamWriter` / `FinishCSVStreamWriter` /
  `CSVStreamWriterDestReceiver` — the client-side primitive that
  opens a libpq COPY-IN to pgduck_server's RECEIVE sink, hands back
  a postgres `DestReceiver` that callers drive with rows, and
  finalizes the deferred query when the stream ends.
- `StreamLocalFileToS3` — uses the same RECEIVE protocol to stream
  iceberg metadata files (metadata.json, manifest list, manifest)
  into pgduck for upload. Replaces the file-based path that needed
  pgduck to see those files locally.
- Cooperative wait via `WaitForResult` in the new primitives:
  `WaitLatchOrSocket` + `PQconsumeInput` + `CHECK_FOR_INTERRUPTS`
  loop, so `statement_timeout` / SIGINT / postmaster-death actually
  fire while the backend is waiting on pgduck. The naive
  `PQgetResult` would not.

`pg_lake_table` (the FDW):

- `multi_data_file_dest.c` (the INSERT-side `MultiDataFileUploadDestReceiver`):
  open a `CSVStreamWriter` instead of a CSV temp file when the GUC
  is on. Each rotation opens a new stream; FlushChildDestReceiver
  finalizes it via FinishCSVStreamWriter. The per-rotation writer
  lives in the receiver's existing childContext, which gets reset
  between rotations. The `partition` pointer on each
  `DataFileModification` is deep-copied under `parentContext` before
  that reset (NULL preserved for unpartitioned tables) so downstream
  consumers in `ApplyDataFileModifications` — which runs at PRE_COMMIT
  in a different memory-context lifetime — don't dereference into
  freed memory.
- `pg_lake_table.c` (the UPDATE/DELETE callsites): the
  `deleteStreamWriter` lazy-open in `DeleteSingleRow` opens a new
  per-source-file stream for the position-delete records, and
  `FinishForeignModify` calls `FinishCSVStreamWriter` to seal it.
  The writer is allocated in a dedicated long-lived
  `deleteStreamMemoryContext` (created at create_foreign_modify
  time, sub of the FDW state context) so it survives the executor's
  per-tuple resets — the same lifetime discipline
  multi_data_file_dest.c already uses for its INSERT-side writer.
  The "all rows deleted, drop the deletion file" optimization in
  `WriteDeleteRecord` is gated to the file-based path; for the
  streaming path, the writer keeps accumulating rows and the
  resulting position-delete file covers all rows in the source —
  semantically equivalent under iceberg's position-delete merge.

`pg_lake_copy` (the COPY pushdown):

- `copy.c` / `copy_io.c`: `COPY foreign_table FROM STDIN` opens a
  `CSVStreamWriter` and pumps the client's CopyData straight through
  to pgduck. The non-pushdown path (when COPY can't be pushed down)
  still uses the file-based code.

Memory-context discipline:

The streaming writers live across multiple rows and sometimes across
the executor's per-tuple ExprContext reset. Each writer is allocated
in a long-lived context dedicated to that writer's lifetime — for
the INSERT side, the existing `MultiDataFileUploadDestReceiver`
childContext; for the DELETE side, a new
`deleteStreamMemoryContext` on the modify state. This mirrors how
the file-based path already explicitly allocates its
`CreateCSVDestReceiver` "in a long-lived memory context" (per the
comment in create_foreign_modify).

Compatibility:

- Default-off GUC: existing deployments are untouched. No behavior
  change unless `streaming_writes = on`.
- The file-based code paths are unchanged. No regressions on
  shared-filesystem topologies.
- No SQL surface changes. No changes to data layout or iceberg
  catalog representation.

Testing:

Verified end-to-end against a CDC-shape workload (200 mixed
INSERT/UPDATE/DELETE events with COMMIT per event) via a local kind
cluster + a GKE deployment with pgduck_server running in a separate
pod from postgres. 10 back-to-back 600s soak runs with full
diagnostic instrumentation across the entire writer lifetime
detected zero correctness issues.

Signed-off-by: Tim McLaughlin <tim@gotab.io>

Author: timmclaughlin

pg_lake_copy/include/pg_lake/copy/copy_io.h

@@ -18,8 +18,11 @@
 #ifndef PG_LAKE_COPY_IO_H
 #define PG_LAKE_COPY_IO_H
 
+#include "libpq-fe.h"
+
 
 void		CopyInputToFile(char *filePath, int columnCount, bool isBinary);
+void		CopyInputToStream(PGconn *streamConn, int columnCount, bool isBinary);
 void		CopyFileToOutput(char *filePath, int columnCount, bool isBinary);
 
 

pg_lake_copy/src/copy/copy.c

@@ -96,6 +96,8 @@ PgLakeCopyValidityCheckHookType PgLakeCopyValidityCheckHook = NULL;
 static bool IsPgLakeCopy(CopyStmt *copyStmt);
 static bool IsCopyFromStdin(CopyStmt *copyStmt);
 static bool IsCopyToStdout(CopyStmt *copyStmt);
+static int64 StreamingCopyFromStdinPushdown(Oid relationId, char *readQuery,
+											TupleDesc tupleDesc);
 static void ProcessPgLakeCopyFrom(CopyStmt *copyStmt, ParseState *pstate,
 								  Relation relation, Node *whereClause,
 								  uint64 *rowsProcessed);
@@ -449,23 +451,52 @@ ProcessPgLakeCopyFrom(CopyStmt *copyStmt, ParseState *pstate, Relation relation,
 	 */
 	TupleDesc	tupleDesc = BuildTupleDescriptorForRelation(relation, copyStmt->attlist);
 
+	/*
+	 * Compute doCopyPushdown up front so the STDIN block can choose between
+	 * the streaming-write path (when GUC is on AND we'll push the COPY down
+	 * to pgduck) and the file-based path (everything else). For non-pushdown
+	 * COPYs the COPY query is wrapped in TRANSMIT and rows round-trip back
+	 * through PG; that's a different shape than the deferred-INSERT we use
+	 * for streaming, so streaming + non-pushdown still uses the file-based
+	 * path. Document the gap.
+	 */
+	bool		doCopyPushdown = IsCopyFromPushdownable(relation, copyStmt->attlist,
+														whereClause, sourceFormat);
+	bool		useStreamingStdin = StreamingWritesEnabled && IsCopyFromStdin(copyStmt) &&
+		doCopyPushdown;
+
 	if (IsCopyFromStdin(copyStmt))
 	{
-		sourcePath = GenerateTempFileName(TEMP_FILE_PATTERN, ensureCleanup);
+		if (useStreamingStdin)
+		{
+			/*
+			 * Streaming-write path: don't park bytes locally. Use the
+			 * pgduck_server RECEIVE sink-path placeholder; the deferred COPY
+			 * query that AddQueryResultToTableStream builds will pick this up
+			 * and substitute it with the server-local sink path before
+			 * running read_csv() on it.
+			 */
+			sourcePath = pstrdup(PG_LAKE_RECV_PATH_PLACEHOLDER);
+		}
+		else
+		{
+			sourcePath = GenerateTempFileName(TEMP_FILE_PATTERN, ensureCleanup);
 
-		/*
-		 * we send the expected column count to make pedantic clients happy
-		 */
-		int			columnCount = tupleDesc->natts;
+			/*
+			 * we send the expected column count to make pedantic clients
+			 * happy
+			 */
+			int			columnCount = tupleDesc->natts;
 
-		bool		isBinary = true;
+			bool		isBinary = true;
 
-		/*
-		 * We copy the incoming bytes to a file first and then try to convert
-		 * that file. We could perhaps optimize this in the future by copying
-		 * via a named pipe.
-		 */
-		CopyInputToFile(sourcePath, columnCount, isBinary);
+			/*
+			 * We copy the incoming bytes to a file first and then try to
+			 * convert that file. We could perhaps optimize this in the future
+			 * by copying via a named pipe.
+			 */
+			CopyInputToFile(sourcePath, columnCount, isBinary);
+		}
 	}
 
 	/*
@@ -483,9 +514,6 @@ ProcessPgLakeCopyFrom(CopyStmt *copyStmt, ParseState *pstate, Relation relation,
 	 * case. However, we do want explicit casts to avoid writing incorrect
 	 * types.
 	 */
-	bool		doCopyPushdown = IsCopyFromPushdownable(relation, copyStmt->attlist,
-														whereClause, sourceFormat);
-
 	if (!doCopyPushdown)
 		readFlags |= READ_DATA_TRANSMIT;
 	else
@@ -535,6 +563,13 @@ ProcessPgLakeCopyFrom(CopyStmt *copyStmt, ParseState *pstate, Relation relation,
 	 */
 	if (doCopyPushdown)
 	{
+		if (useStreamingStdin)
+		{
+			*rowsProcessed = StreamingCopyFromStdinPushdown(relationId, readQuery,
+															tupleDesc);
+			return;
+		}
+
 		*rowsProcessed = AddQueryResultToTable(relationId, readQuery, tupleDesc, true);
 		return;
 	}
@@ -660,6 +695,38 @@ IsCopyFromPushdownable(Relation relation, List *columnNameList,
 }
 
 
+/*
+ * StreamingCopyFromStdinPushdown drives the COPY tablename FROM STDIN
+ * pushdown path when pg_lake_engine.streaming_writes=on.
+ *
+ * Pulled out of ProcessPgLakeCopyFrom so the caller stays small enough
+ * for gcc's clobbered-variable analysis (-Wclobbered) to keep its
+ * pre-streaming behavior on the unrelated TupleDesc helpers below;
+ * inlining a multi-step PG-call sequence into ProcessPgLakeCopyFrom
+ * triggered false-positive clobber warnings on attributeDescriptor /
+ * cleanTupleDesc that don't fire when this branch is its own function.
+ *
+ * No PG_TRY/CATCH: errors here trigger transaction abort; pgduck's
+ * PGDuckClientTransactionCallback recycles the libpq connection and
+ * the in-progress-file record registered inside Start... is cleaned
+ * up by the pre-commit hook on abort.
+ */
+static int64
+StreamingCopyFromStdinPushdown(Oid relationId, char *readQuery, TupleDesc tupleDesc)
+{
+	AddQueryResultStreamHandle *handle =
+		StartAddQueryResultToTableStream(relationId, readQuery, tupleDesc,
+										  /* wrapNativeTypes */ true);
+
+	int			columnCount = tupleDesc->natts;
+	bool		isBinary = true;
+
+	CopyInputToStream(AddQueryResultStreamConnection(handle), columnCount, isBinary);
+
+	return FinishAddQueryResultToTableStream(handle);
+}
+
+
 /*
 * GenerateReadDataSourceQuery generates the pgduck query to read the given
 * source format/path.

pg_lake_copy/src/copy/copy_io.c

@@ -30,6 +30,7 @@
 #include "storage/fd.h"
 #include "utils/memutils.h"
 
+#include "libpq-fe.h"
 #include "libpq/libpq.h"
 
 #define MAX_READ_SIZE (65536)
@@ -55,6 +56,55 @@ static void SendCopyEnd(void);
 static void SendCopyData(char *sendBuffer, int sendBufferLength);
 
 
+/*
+ * CopyInputToStream is the streaming counterpart of CopyInputToFile.
+ *
+ * Instead of writing the postgres client's COPY-IN bytes to a local file,
+ * forwards them via PQputCopyData on `streamConn`. The caller must have
+ * already opened a libpq COPY-IN stream against pgduck_server (via
+ * OpenCopyInStreamToPGDuck) so streamConn is in COPY-IN-active state.
+ * Uses the same SendCopyInResponseToClient(columnCount, isBinary) the
+ * file-based path uses to tell the client we're ready.
+ *
+ * Caller is responsible for finalizing the libpq stream via
+ * FinishCopyInStreamToPGDuck (or moral equivalent) afterwards.
+ */
+void
+CopyInputToStream(PGconn *streamConn, int columnCount, bool isBinary)
+{
+	CopyFromStdinState cstate = {
+		.fe_msgbuf = makeStringInfo(),
+		.raw_reached_eof = false
+	};
+
+	/* tell the client we are ready for data */
+	SendCopyInResponseToClient(columnCount, isBinary);
+
+	char	   *receiveBuffer = palloc(MAX_READ_SIZE);
+
+	while (!cstate.raw_reached_eof)
+	{
+		unsigned long bytesRead = ReceiveDataFromClient(&cstate, receiveBuffer);
+
+		if (bytesRead > 0)
+		{
+			if (PQputCopyData(streamConn, receiveBuffer, (int) bytesRead) != 1)
+			{
+				ereport(ERROR, (errcode(ERRCODE_CONNECTION_FAILURE),
+								errmsg("failed to forward COPY data to pgduck_server: %s",
+									   PQerrorMessage(streamConn))));
+			}
+		}
+		else if (bytesRead == 0)
+		{
+			break;
+		}
+	}
+
+	pfree(receiveBuffer);
+}
+
+
 /*
  * CopyInputToFile copies data from the socket to the given file.
  * We request the client send a specific column count.

pg_lake_engine/include/pg_lake/csv/csv_writer.h

@@ -18,6 +18,8 @@
 #ifndef CSV_WRITER_H
 #define CSV_WRITER_H
 
+#include "libpq-fe.h"
+
 #include "pg_lake/copy/copy_format.h"
 #include "tcop/dest.h"
 #include "nodes/pg_list.h"
@@ -28,6 +30,22 @@ extern PGDLLEXPORT DestReceiver *CreateCSVDestReceiverExtended(char *filename,
 															   List *copyOptions,
 															   CopyDataFormat targetFormat,
 															   bool sessionLifetime);
+
+/*
+ * Streaming variant: bytes go to a libpq COPY-IN already opened on
+ * `streamConn` (use OpenCopyInStreamToPGDuck() to get there). The caller
+ * is responsible for finalizing the stream via FinishCopyInStreamToPGDuck()
+ * AFTER calling rShutdown on the returned DestReceiver — rShutdown only
+ * flushes the per-row buffer and emits any binary trailer; it deliberately
+ * does NOT call PQputCopyEnd, so callers can still emit additional
+ * CopyData (e.g. for multi-segment writes) before closing. The deferred
+ * query's PGresult is returned by FinishCopyInStreamToPGDuck so callers
+ * can extract row counts / column statistics.
+ */
+extern PGDLLEXPORT DestReceiver *CreateCSVStreamDestReceiver(PGconn *streamConn,
+															 List *copyOptions,
+															 CopyDataFormat targetFormat);
+
 extern PGDLLEXPORT int GetCSVDestReceiverMaxLineSize(DestReceiver *dest);
 extern PGDLLEXPORT uint64 GetCSVDestReceiverFileSize(DestReceiver *dest);
 

pg_lake_engine/include/pg_lake/pgduck/client.h

@@ -52,4 +52,27 @@ extern PGDLLEXPORT char *GetSingleValueFromPGDuck(char *query);
 extern PGDLLEXPORT void SendQueryWithParams(PGDuckConnection * pgduckConn, char *queryString,
 											int numParams, const char **parameterValues);
 
+/*
+ * Streaming-write helpers for the RECEIVE protocol prefix on pgduck_server.
+ *
+ * OpenCopyInStreamToPGDuck sends `queryString` (which must begin with
+ * "RECEIVE " and contain '@@PG_LAKE_RECV@@' as the read_csv path
+ * placeholder) and waits for pgduck_server's CopyInResponse. After this
+ * returns, the caller may emit CSV bytes via PQputCopyData on
+ * pgDuckConnection->conn (typically by passing it to
+ * CreateCSVStreamDestReceiver and driving a producer query through it).
+ *
+ * FinishCopyInStreamToPGDuck calls PQputCopyEnd(NULL), waits for the
+ * deferred query's first PGresult, drains the trailing NULL terminator,
+ * and returns the PGresult to the caller. The caller owns it and must
+ * PQclear it. Errors raised by the deferred query (e.g. INSERT failures)
+ * surface here as ereport(ERROR) before returning.
+ *
+ * Both throw on protocol or query errors; the connection is left in an
+ * idle state on success, in an error state on failure.
+ */
+extern PGDLLEXPORT void OpenCopyInStreamToPGDuck(PGDuckConnection * pgDuckConnection,
+												 const char *queryString);
+extern PGDLLEXPORT PGresult *FinishCopyInStreamToPGDuck(PGDuckConnection * pgDuckConnection);
+
 #endif

pg_lake_engine/include/pg_lake/pgduck/write_data.h

@@ -23,11 +23,25 @@
 #include "pg_lake/parquet/field.h"
 #include "pg_lake/pgduck/iceberg_validation.h"
 #include "nodes/pg_list.h"
+#include "tcop/dest.h"
 
 /* pg_lake_table.target_row_group_size_mb */
 #define DEFAULT_TARGET_ROW_GROUP_SIZE_MB 512
 extern PGDLLEXPORT int TargetRowGroupSizeMB;
 
+/* pg_lake_engine.streaming_writes */
+extern PGDLLEXPORT bool StreamingWritesEnabled;
+
+/*
+ * Sink-path placeholder substituted server-side by pgduck_server's
+ * RECEIVE handler. Must match SINK_PLACEHOLDER in pgduck_server's
+ * pgsession.c. Streaming-write callers pass this where the file-based
+ * path would pass the local CSV file path; the placeholder is then
+ * embedded inside read_csv('<placeholder>', ...) and pgduck swaps in
+ * the server-local sink path before running the deferred query.
+ */
+#define PG_LAKE_RECV_PATH_PLACEHOLDER "@@PG_LAKE_RECV@@"
+
 typedef enum ParquetVersion
 {
 	PARQUET_VERSION_V1 = 1,
@@ -57,7 +71,42 @@ extern PGDLLEXPORT StatsCollector * WriteQueryResultTo(char *query,
 													   List *leafFields,
 													   IcebergOutOfRangePolicy outOfRangePolicy,
 													   bool wrapNativeTypes);
+/*
+ * BuildCopyToCommandString assembles `COPY (query) TO 'destinationPath'
+ * WITH (format ..., compression ..., return_stats, ...)` for the given
+ * destination format / compression / format options. The same string
+ * builder is used by WriteQueryResultTo (file path), OpenCSVStreamWriter
+ * (streaming), and the streaming AddQueryResultToTable variant in
+ * writable_table.c.
+ */
+extern PGDLLEXPORT char *BuildCopyToCommandString(char *query, char *destinationPath,
+												  CopyDataFormat destinationFormat,
+												  CopyDataCompression destinationCompression,
+												  List *formatOptions,
+												  bool queryHasRowId,
+												  DataFileSchema * schema,
+												  TupleDesc queryTupleDesc,
+												  IcebergOutOfRangePolicy outOfRangePolicy,
+												  bool wrapNativeTypes);
+
 extern PGDLLEXPORT void AppendFields(StringInfo map, DataFileSchema * schema);
 extern PGDLLEXPORT char *TupleDescToColumnMapForWrite(TupleDesc tupleDesc, CopyDataFormat destinationFormat);
 extern PGDLLEXPORT char *TupleDescToProjectionListForWrite(TupleDesc tupleDesc,
 														   CopyDataFormat destinationFormat);
+
+/*
+ * Streaming counterpart of ConvertCSVFileTo. See OpenCSVStreamWriter in
+ * write_data.c for the contract. The struct is opaque to callers.
+ */
+typedef struct CSVStreamWriter CSVStreamWriter;
+
+extern PGDLLEXPORT CSVStreamWriter * OpenCSVStreamWriter(TupleDesc csvTupleDesc,
+														 int maxLineSize,
+														 char *destinationPath,
+														 CopyDataFormat destinationFormat,
+														 CopyDataCompression destinationCompression,
+														 List *formatOptions,
+														 DataFileSchema * schema,
+														 List *leafFields);
+extern PGDLLEXPORT DestReceiver *CSVStreamWriterDestReceiver(CSVStreamWriter * writer);
+extern PGDLLEXPORT StatsCollector * FinishCSVStreamWriter(CSVStreamWriter * writer);