pgduck_server: add RECEIVE protocol for streaming COPY-IN sink

Adds a TCP listener and a new "RECEIVE" query prefix to pgduck_server.
With these in place, a remote client can stream CSV (or other data)
to a server-local sink path via the standard libpq COPY-IN protocol,
and pgduck runs a deferred query reading from that path once
CopyDone arrives.

Motivation:

The existing pgduck_server design assumes the client and pgduck share
a filesystem — the standard pg_lake "sidecar" deployment colocates
postgres and pgduck_server in the same pod and lets pg_lake's bulk-
write paths drop CSV under $PGDATA/pgsql_tmp for pgduck to read with
read_csv(). That topology breaks under deployment models that don't
co-locate the two processes (operator-managed Kubernetes deployments
where postgres and pgduck run in separate pods, multi-host setups,
etc.). The streaming-write paths in the companion pg_lake patch lean
on this RECEIVE protocol to push bytes directly to pgduck via libpq
without needing a shared filesystem.

What this patch adds:

- TCP listener on pgduck_server controlled by --listen-addresses /
  --port (default Unix-socket path remains supported). Lets remote
  postgres backends reach pgduck over the network.
- A "RECEIVE <inner-query>" query prefix recognized by
  process_query_message. When it sees "RECEIVE …", the server:
    1. Picks a server-local sink path (under --recv-dir).
    2. Substitutes the bare token "@@PGLAKE_RECV_SINK@@" inside the
       inner query with a properly-quoted SQL literal containing
       the sink path. The server adds the surrounding single quotes
       and escapes any embedded single quotes; the placeholder
       itself is intentionally NOT inside a SQL string literal in
       the client-emitted query, so it can never collide with user-
       supplied data.
    3. Sends CopyInResponse to the client and accepts CopyData
       chunks, writing them straight to the sink.
    4. On CopyDone, runs the deferred inner query against the sink
       path and streams its result rows back via the standard
       PGresult flow.
  This lets clients use the existing libpq COPY-IN flow as the
  transport for arbitrary inner queries that read from a path.
- recv_sink module: opens, writes, and cleans up the per-client sink
  files; bounded by --recv-max-bytes; refuses path traversal.
- SSL-negotiate-byte handling: pgduck_server replies 'N' to libpq's
  SSLRequest instead of closing the connection. Lets clients with
  sslmode=prefer fall back to plaintext cleanly.
- Explicit pgsession_flush() after pgsession_send_copy_in_response.
  Without this, the 5-byte CopyInResponse can sit in pgduck's send
  buffer indefinitely (no auto-flush on small messages); the client
  blocks forever in PQputCopyData waiting for the server to be ready,
  and the RECEIVE handshake deadlocks. Found while debugging exactly
  that ~4-hour hang during smoke runs.

Compatibility:

- Existing simple SELECT and COPY paths are unchanged.
- Unix-socket transport remains supported when --listen-addresses is
  not set; nothing forces TCP.
- No new dependencies. The recv_sink uses the same memory and I/O
  primitives the rest of pgsession.c uses.
- The "RECEIVE" prefix is only recognized in the simple-query path
  (because it needs the server-side prefix detection that the
  extended-query protocol doesn't offer). Clients using
  PQsendQueryParams keep using the existing path.

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

Author: timmclaughlin

pgduck_server/include/command_line/command_line.h

@@ -33,6 +33,16 @@ typedef struct
 	char	   *unix_socket_directory;
 	char	   *unix_socket_group;
 	int			unix_socket_permissions;
+
+	/*
+	 * Comma-separated list of TCP addresses to bind on (PostgreSQL-style
+	 * semantics). Empty string or NULL means TCP is disabled (default) and
+	 * pgduck_server only listens on the Unix domain socket. Examples:
+	 * "0.0.0.0"     listen on all IPv4 interfaces "0.0.0.0,::"  listen on all
+	 * IPv4 + all IPv6 interfaces "127.0.0.1"   loopback only The TCP port is
+	 * the same `port` field used for the Unix socket suffix.
+	 */
+	char	   *listen_addresses;
 	unsigned int port;
 	unsigned int max_clients;
 	char	   *memory_limit;
@@ -46,6 +56,12 @@ typedef struct
 	bool		debug;
 	char	   *init_file_path;
 	char	   *pidfile_path;
+
+	/*
+	 * Base directory for RECEIVE-query sinks (libpq COPY-IN landing files).
+	 * Defaults to <cache_dir>/recv when unset; overridable via --recv-dir.
+	 */
+	char	   *recv_dir;
 }			CommandLineOptions;
 
 CommandLineOptions parse_arguments(int argc, char *argv[]);

pgduck_server/include/pgserver/pgserver.h

@@ -23,14 +23,31 @@
 #ifndef PGDUCK_PG_SERVER_H
 #define PGDUCK_PG_SERVER_H
 
+/*
+ * Maximum number of TCP listening sockets we'll bind. Each address in
+ * --listen_addresses can resolve to multiple addrinfos (rare; usually
+ * one per family), so we size for headroom over typical configs like
+ * "0.0.0.0,::".
+ */
+#define MAX_TCP_LISTEN_SOCKETS 16
+
 /*
  * PGServer represents an instance of a PostgreSQL wire compatible
  * server.
  */
 typedef struct PGServer
 {
 	int			listeningPort;
-	int			listeningSocket;
+	int			listeningSocket;	/* Unix domain socket */
+
+	/*
+	 * TCP listening sockets (one per resolved address from
+	 * --listen_addresses). numTcpSockets == 0 when TCP is disabled (the
+	 * default), in which case the server only accepts Unix-socket connections —
+	 * preserving backwards-compatible behavior.
+	 */
+	int			tcpSockets[MAX_TCP_LISTEN_SOCKETS];
+	int			numTcpSockets;
 
 	char		unixSocketDir[MAXPGPATH];
 	char		unixSocketPath[MAXPGPATH];
@@ -48,6 +65,7 @@ extern int	pgserver_init(PGServer * pgServer,
 						  char *unixSocketPath,
 						  char *unixSocketOwningGroup,
 						  int unixSocketPermissions,
+						  char *tcpListenAddresses,
 						  int port);
 extern int	pgserver_run(PGServer * pgServer);
 extern void pgserver_destroy(PGServer * pgServer);

pgduck_server/include/pgsession/pgsession.h

@@ -28,6 +28,9 @@
 
 #include "duckdb/duckdb.h"
 
+/* forward declaration; full definition lives in pgsession/recv_sink.h */
+typedef struct ReceiveSink ReceiveSink;
+
 #define DUCKPG_SERVER_VERSION "16.4.DuckPG"
 
 /*
@@ -175,6 +178,19 @@ typedef struct PGSession
 
 	int			lastReportedSendErrno;	/* needed to make pgsession_flush
 										 * thread-safe */
+
+	/*
+	 * RECEIVE-query state.
+	 *
+	 * When a query begins with the "RECEIVE " prefix, we send the client a
+	 * CopyInResponse, accept CopyData messages into activeRecvSink, and defer
+	 * running the actual DuckDB query until CopyDone arrives. While
+	 * deferredQueryString is non-NULL the session is in COPY-IN-active state
+	 * and only 'd' / 'c' / 'f' messages are valid.
+	 */
+	ReceiveSink *activeRecvSink;
+	char	   *deferredQueryString;
+	ResponseFormat deferredResponseFormat;
 }			PGSession;
 
 /* per-client entrance point for the pgsession logic */

pgduck_server/include/pgsession/recv_sink.h

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2026 Snowflake Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/*
+ * Receive sink: a per-RECEIVE-query landing area on pgduck_server's local
+ * filesystem.
+ *
+ * The RECEIVE query prefix (mirror of TRANSMIT) lets pg_lake clients stream
+ * bytes via libpq COPY IN, instead of writing to a shared filesystem under
+ * the Postgres backend's PGDATA. The bytes land in a sink owned by
+ * pgduck_server, and the sink path is what gets passed to DuckDB's
+ * read_csv() etc. This decouples pgduck_server's filesystem from the
+ * client's PGDATA, which is a hard requirement when pgduck_server runs in
+ * a different pod/host than the Postgres backend (e.g. on Kubernetes).
+ *
+ * v1 stages received bytes to a regular file in a private base directory.
+ * Future versions may use a FIFO so DuckDB can consume the stream in
+ * parallel with the client upload, but the file approach is simpler and
+ * sufficient for typical CDC batch sizes (tens of MB).
+ */
+#ifndef PGDUCK_RECV_SINK_H
+#define PGDUCK_RECV_SINK_H
+
+#include "c.h"
+#include <stddef.h>
+
+typedef struct ReceiveSink ReceiveSink;
+
+/*
+ * Initialize the global receive directory. Creates it (mode 0700) if
+ * missing and unlinks any stale files left over from a prior run. Must be
+ * called once at startup, before any sink is created. Repeated calls
+ * replace the configured path.
+ *
+ * Returns 0 on success, -1 on error (with errno set and the failure
+ * already logged).
+ */
+int			recv_sink_global_init(const char *base_dir);
+
+/*
+ * Returns the configured base directory, or NULL if recv_sink has not been
+ * initialized.
+ */
+const char *recv_sink_global_dir(void);
+
+/*
+ * Create a new sink. Allocates a unique path under the global recv
+ * directory, opens it for writing (mode 0600), and returns the sink. The
+ * caller owns the sink and must call recv_sink_destroy() when done.
+ *
+ * Returns NULL on error (with errno set).
+ */
+ReceiveSink *recv_sink_create(void);
+
+/*
+ * Path of the sink file. Stable from create() through destroy(). DuckDB
+ * should open this path for reading.
+ */
+const char *recv_sink_path(const ReceiveSink * sink);
+
+/*
+ * Append bytes to the sink. Performs a full write (handles partial writes
+ * and EINTR). Returns 0 on success, -1 on error.
+ */
+int			recv_sink_write(ReceiveSink * sink, const char *data, size_t len);
+
+/*
+ * Close the sink for writing. After this call, the sink path is readable
+ * by DuckDB and reads will see EOF after all written bytes. Idempotent.
+ * Returns 0 on success, -1 on error.
+ */
+int			recv_sink_finalize(ReceiveSink * sink);
+
+/*
+ * Destroy the sink: close the file if still open, unlink the path, and
+ * free memory. Always succeeds; safe to call with NULL.
+ */
+void		recv_sink_destroy(ReceiveSink * sink);
+
+#endif							/* PGDUCK_RECV_SINK_H */

pgduck_server/src/command_line/command_line.c

@@ -57,7 +57,8 @@ print_usage()
 	printf(" --unix_socket_directory <path>		Specify the unix socket directory, default is %s\n", DEFAULT_UNIX_DOMAIN_PATH);
 	printf(" --unix_socket_group <group name>	Specify the unix socket group owner, default is \"%s\"\n", DEFAULT_UNIX_DOMAIN_GROUP);
 	printf(" --unix_socket_permissions <mask>	Specify the unix socket (chmod) permissions, default is %o\n", DEFAULT_UNIX_DOMAIN_PERMISSIONS);
-	printf(" --port <port>                 		Specify the port number, default is %d\n", DEFAULT_PORT);
+	printf(" --listen_addresses <addrs>		Comma-separated TCP addresses to listen on (e.g. \"0.0.0.0,::\"). Default: empty (Unix socket only)\n");
+	printf(" --port <port>                 		Specify the port number (used for both Unix socket suffix and TCP listener), default is %d\n", DEFAULT_PORT);
 	printf(" --max_clients <max_clients>		Specify the maximum allowed clients, default is %d\n", DEFAULT_MAX_CLIENTS);
 	printf(" --memory_limit=<memory_limit>		Optionally specify the maximum memory of pgduck_server similar to DuckDB's memory_limit, the default is 80 percent of the system memory\n");
 	printf(" --continue_on_oom                  If out of memory error occurs, continue operating\n");
@@ -66,6 +67,7 @@ print_usage()
 	printf(" --check_cli_params_only       		Only check the cli arguments, do not run the server\n");
 	printf(" --init_file_path <path>			Execute all statements in this file on start-up\n");
 	printf(" --cache_dir                    	Specify the directory to use to cache remote files (from S3)\n");
+	printf(" --recv_dir <path>                  Base dir for RECEIVE-query landing files; default <cache_dir>/recv or /tmp/pgduck_recv\n");
 	printf(" --extensions_dir <path>			Install and load extensions in the specified directory\n");
 	printf(" --pidfile <path>					Write the pid of this program to the given path\n");
 	printf(" --no_extension_install             Disable extension installation\n");
@@ -85,6 +87,7 @@ parse_arguments(int argc, char *argv[])
 		.unix_socket_directory = DEFAULT_UNIX_DOMAIN_PATH,
 		.unix_socket_group = DEFAULT_UNIX_DOMAIN_GROUP,
 		.unix_socket_permissions = DEFAULT_UNIX_DOMAIN_PERMISSIONS,
+		.listen_addresses = NULL,
 		.port = DEFAULT_PORT,
 		.max_clients = DEFAULT_MAX_CLIENTS,
 		.memory_limit = NULL,
@@ -97,6 +100,7 @@ parse_arguments(int argc, char *argv[])
 		.extensions_dir = NULL,
 		.no_extension_install = false,
 		.debug = false,
+		.recv_dir = NULL,
 	};
 	int			opt;
 	int			option_index = 0;
@@ -108,6 +112,7 @@ parse_arguments(int argc, char *argv[])
 		{"unix_socket_directory", required_argument, NULL, 'U'},
 		{"unix_socket_group", required_argument, NULL, 'G'},
 		{"unix_socket_permissions", required_argument, NULL, 'm'},
+		{"listen_addresses", required_argument, NULL, 'A'},
 		{"port", required_argument, NULL, 'P'},
 		{"max_clients", required_argument, NULL, 'M'},
 		{"memory_limit", required_argument, NULL, 'l'},
@@ -120,10 +125,11 @@ parse_arguments(int argc, char *argv[])
 		{"init_file_path", required_argument, NULL, 'i'},
 		{"pidfile", required_argument, NULL, 'p'},
 		{"debug", no_argument, NULL, 'd'},
+		{"recv_dir", required_argument, NULL, 'R'},
 		{0, 0, 0, 0}
 	};
 
-	while ((opt = getopt_long(argc, argv, "cvhU:P:M:D:l:L:p:d", long_options, &option_index)) != -1)
+	while ((opt = getopt_long(argc, argv, "cvhU:A:P:M:D:l:L:p:dR:", long_options, &option_index)) != -1)
 	{
 		switch (opt)
 		{
@@ -143,6 +149,15 @@ parse_arguments(int argc, char *argv[])
 			case 'G':
 				options.unix_socket_group = strdup(optarg);
 				break;
+			case 'A':
+
+				/*
+				 * Empty string is allowed and disables TCP listening (same as
+				 * not passing the flag at all). Any non-empty value is
+				 * treated as a comma-separated list of addresses to bind on.
+				 */
+				options.listen_addresses = strdup(optarg);
+				break;
 			case 'l':
 				if (optarg)
 					options.memory_limit = strdup(optarg);
@@ -252,6 +267,9 @@ parse_arguments(int argc, char *argv[])
 			case 'p':
 				options.pidfile_path = strdup(optarg);
 				break;
+			case 'R':
+				options.recv_dir = strdup(optarg);
+				break;
 			case '?':
 				print_usage();
 				exit(EXIT_FAILURE);

pgduck_server/src/main.c

@@ -22,6 +22,8 @@
  */
 #include <stdio.h>
 #include <signal.h>
+#include <stdlib.h>
+#include <string.h>
 
 #include "c.h"
 #include "postgres_fe.h"
@@ -32,6 +34,7 @@
 #include "pgserver/pgserver.h"
 #include "pgserver/client_threadpool.h"
 #include "pgsession/pgsession.h"
+#include "pgsession/recv_sink.h"
 #include "duckdb/duckdb.h"
 #include "utils/pidfile.h"
 
@@ -74,6 +77,47 @@ main(int argc, char *argv[])
 
 	pgclient_threadpool_init(options.max_clients);
 
+	/*
+	 * Resolve and initialize the RECEIVE-query landing directory.
+	 *
+	 * Default precedence: --recv_dir > <cache_dir>/recv > /tmp/pgduck_recv.
+	 * The directory is created (mode 0700) and any stale files from a prior
+	 * run are unlinked.
+	 */
+	{
+		char	   *recv_dir = options.recv_dir;
+		char	   *resolved = NULL;
+
+		if (recv_dir == NULL)
+		{
+			if (options.cache_dir != NULL)
+			{
+				size_t		len = strlen(options.cache_dir) + sizeof("/recv");
+
+				resolved = malloc(len);
+				if (resolved == NULL)
+				{
+					PGDUCK_SERVER_ERROR("out of memory deriving recv_dir");
+					return STATUS_ERROR;
+				}
+				snprintf(resolved, len, "%s/recv", options.cache_dir);
+				recv_dir = resolved;
+			}
+			else
+			{
+				recv_dir = "/tmp/pgduck_recv";
+			}
+		}
+
+		if (recv_sink_global_init(recv_dir) != 0)
+		{
+			free(resolved);
+			return STATUS_ERROR;
+		}
+		PGDUCK_SERVER_LOG("RECEIVE-query landing dir: %s", recv_dir);
+		free(resolved);
+	}
+
 	PGServer	pgServer;
 
 	srand(time(NULL));
@@ -82,6 +126,7 @@ main(int argc, char *argv[])
 					  options.unix_socket_directory,
 					  options.unix_socket_group,
 					  options.unix_socket_permissions,
+					  options.listen_addresses,
 					  options.port) != STATUS_OK)
 		return STATUS_ERROR;