Support streaming COPY FROM STDIN and COPY TO STDOUT

vertx-pg-client/src/main/asciidoc/index.adoc

@@ -13,6 +13,7 @@ The client is reactive and non-blocking, allowing to handle many database connec
 * Publish / subscribe using PostgreSQL `NOTIFY/LISTEN`
 * Batch and cursor
 * Row streaming
+* `COPY` streaming
 * Command pipelining
 * RxJava API
 * Direct memory to object without unnecessary copies
@@ -656,6 +657,84 @@ create easily create a string directly from the row set:
 {@link examples.PgClientExamples#collector02Example}
 ----
 
+== Streaming COPY
+
+PostgreSQL `COPY` transfers bulk data between a table and a stream, without going through
+the extended query protocol. The client exposes both directions as raw byte streams, so the
+data never has to be materialized in memory.
+
+Each call accepts exactly one `COPY` statement, and a connection can have a single COPY
+operation in progress at a time.
+
+=== COPY TO STDOUT
+
+{@link io.vertx.pgclient.PgConnection#copyOut(java.lang.String)} executes a
+`COPY ... TO STDOUT` statement and returns a {@link io.vertx.pgclient.PgCopyOut}, a
+`ReadStream` of `Buffer`. The future is notified once the server has entered COPY OUT mode.
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyOut(io.vertx.pgclient.PgConnection)}
+----
+
+{@link io.vertx.pgclient.PgCopyOut#completion()} is notified with the number of rows
+reported by PostgreSQL.
+
+The stream is _hot_: like other Vert.x read streams it starts in flowing mode. Consuming it
+is what advances the connection, so `completion()` remains pending for as long as the stream
+is paused or has no data handler. Call `pause()` before installing a handler when you need
+explicit demand control:
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyOutPaused(io.vertx.pgclient.PgConnection)}
+----
+
+By default each PostgreSQL `CopyData` message is emitted as one buffer. When message
+boundaries do not matter and throughput does, use
+{@link io.vertx.pgclient.PgCopyOutOptions#setAggregationThreshold(int)} to combine small
+messages into larger buffers:
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyOutWithOptions(io.vertx.pgclient.PgConnection)}
+----
+
+=== COPY FROM STDIN
+
+{@link io.vertx.pgclient.PgConnection#copyIn(java.lang.String)} executes a
+`COPY ... FROM STDIN` statement and returns a {@link io.vertx.pgclient.PgCopyIn}, a
+`WriteStream` of `Buffer`. Write the data, then call `end()` to send `CopyDone`:
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyIn(io.vertx.pgclient.PgConnection)}
+----
+
+The future returned by `end()` is notified when the COPY command itself completes, not
+merely when the last buffer has been written. The row count is available from
+{@link io.vertx.pgclient.PgCopyIn#completion()}.
+
+The stream honours `writeQueueFull()` and `drainHandler()` like any other `WriteStream`, so
+it can be fed with a `Pipe` or with manual flow control.
+
+Small writes are combined so that they do not become many small messages on the wire. Use
+{@link io.vertx.pgclient.PgCopyInOptions#setChunkSize(int)} to change how much is combined
+before a payload is sent; a write larger than that is sent on its own:
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyInWithOptions(io.vertx.pgclient.PgConnection)}
+----
+
+To give up on a copy in progress, call {@link io.vertx.pgclient.PgCopyIn#abort(java.lang.String)}.
+It sends a `CopyFail` message and PostgreSQL rolls the copy back:
+
+[source,$lang]
+----
+{@link examples.PgClientExamples#copyInAbort(io.vertx.pgclient.PgConnection)}
+----
+
 == Pub/sub
 
 PostgreSQL supports pub/sub communication channels.

vertx-pg-client/src/main/java/examples/PgClientExamples.java

@@ -13,6 +13,7 @@
 
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.ClientSSLOptions;
 import io.vertx.core.net.PemTrustOptions;
@@ -824,4 +825,89 @@ public void batchReturning(SqlClient client) {
   public void pgBouncer(PgConnectOptions connectOptions) {
     connectOptions.setUseLayer7Proxy(true);
   }
+
+  public void copyOut(PgConnection connection) {
+    connection
+      .copyOut("COPY users TO STDOUT (FORMAT csv)")
+      .onSuccess(stream -> {
+        stream.handler(buffer -> {
+          System.out.println("Received " + buffer.length() + " bytes");
+        });
+        stream.completion().onSuccess(rows -> {
+          System.out.println("Copied " + rows + " rows");
+        });
+      });
+  }
+
+  public void copyOutPaused(PgConnection connection) {
+    connection
+      .copyOut("COPY users TO STDOUT (FORMAT csv)")
+      .onSuccess(stream -> {
+
+        // Take control of the demand before any data is emitted
+        stream.pause();
+
+        stream.handler(buffer -> {
+          System.out.println("Received " + buffer.length() + " bytes");
+        });
+
+        // Ask for the next chunk when you are ready for it
+        stream.fetch(1);
+      });
+  }
+
+  public void copyOutWithOptions(PgConnection connection) {
+    PgCopyOutOptions options = new PgCopyOutOptions()
+      .setAggregationThreshold(64 * 1024);
+
+    connection
+      .copyOut("COPY users TO STDOUT (FORMAT csv)", options)
+      .onSuccess(stream -> {
+        stream.handler(buffer -> {
+          System.out.println("Received " + buffer.length() + " bytes");
+        });
+      });
+  }
+
+  public void copyIn(PgConnection connection) {
+    connection
+      .copyIn("COPY users FROM STDIN (FORMAT csv)")
+      .onSuccess(stream -> {
+
+        stream.write(Buffer.buffer("1,Julien\n"));
+        stream.write(Buffer.buffer("2,Emad\n"));
+
+        // Sends CopyDone, the future is notified when the COPY command completes
+        stream.end().onSuccess(v -> {
+          System.out.println("Copy completed");
+        });
+
+        stream.completion().onSuccess(rows -> {
+          System.out.println("Copied " + rows + " rows");
+        });
+      });
+  }
+
+  public void copyInWithOptions(PgConnection connection) {
+    PgCopyInOptions options = new PgCopyInOptions()
+      .setChunkSize(512 * 1024);
+
+    connection
+      .copyIn("COPY users FROM STDIN (FORMAT csv)", options)
+      .onSuccess(stream -> {
+        stream.write(Buffer.buffer("1,Julien\n"));
+        stream.end();
+      });
+  }
+
+  public void copyInAbort(PgConnection connection) {
+    connection
+      .copyIn("COPY users FROM STDIN (FORMAT csv)")
+      .onSuccess(stream -> {
+        stream.write(Buffer.buffer("1,Julien\n"));
+
+        // Sends CopyFail, the server rolls the COPY back
+        stream.abort("aborted by the application");
+      });
+  }
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/PgConnection.java

@@ -135,4 +135,48 @@ static Future<PgConnection> connect(Vertx vertx, String connectionUri) {
   static PgConnection cast(SqlConnection sqlConnection) {
     return (PgConnection) sqlConnection;
   }
+
+  /**
+   * Execute a {@code COPY ... TO STDOUT} statement and stream the raw data produced by the server.
+   * <p/>
+   * The returned future is notified when PostgreSQL has entered COPY OUT mode, with a
+   * {@link PgCopyOut} emitting the copy data as {@link io.vertx.core.buffer.Buffer} chunks.
+   * A connection can have a single COPY operation in progress at a time.
+   *
+   * @param sql a single {@code COPY ... TO STDOUT} statement
+   * @return a future notified with the stream or the failure
+   */
+  Future<PgCopyOut> copyOut(String sql);
+
+  /**
+   * Like {@link #copyOut(String)} but with the given {@code options}.
+   *
+   * @param sql a single {@code COPY ... TO STDOUT} statement
+   * @param options the copy out options
+   * @return a future notified with the stream or the failure
+   */
+  Future<PgCopyOut> copyOut(String sql, PgCopyOutOptions options);
+
+  /**
+   * Execute a {@code COPY ... FROM STDIN} statement and stream raw data to the server.
+   * <p/>
+   * The returned future is notified when PostgreSQL has entered COPY IN mode, with a
+   * {@link PgCopyIn} accepting the copy data as {@link io.vertx.core.buffer.Buffer} chunks.
+   * Call {@link PgCopyIn#end()} to signal the end of the data, or
+   * {@link PgCopyIn#abort(String)} to roll the copy back. A connection can have a single
+   * COPY operation in progress at a time.
+   *
+   * @param sql a single {@code COPY ... FROM STDIN} statement
+   * @return a future notified with the stream or the failure
+   */
+  Future<PgCopyIn> copyIn(String sql);
+
+  /**
+   * Like {@link #copyIn(String)} but with the given {@code options}.
+   *
+   * @param sql a single {@code COPY ... FROM STDIN} statement
+   * @param options the copy in options
+   * @return a future notified with the stream or the failure
+   */
+  Future<PgCopyIn> copyIn(String sql, PgCopyInOptions options);
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/PgCopyIn.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2011-2026 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.pgclient;
+
+import io.vertx.codegen.annotations.Fluent;
+import io.vertx.codegen.annotations.VertxGen;
+import io.vertx.core.Future;
+import io.vertx.core.Handler;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.core.streams.WriteStream;
+
+/**
+ * COPY FROM STDIN sink as a Vert.x WriteStream.
+ *
+ * The stream accepts Buffer chunks and sends them as CopyData messages.
+ * {@link #completion()} completes with the COPY row count when the server completes the COPY command.
+ */
+@VertxGen(concrete = false)
+public interface PgCopyIn extends WriteStream<Buffer> {
+
+  @Fluent
+  @Override
+  PgCopyIn exceptionHandler(Handler<Throwable> handler);
+
+  @Fluent
+  @Override
+  PgCopyIn setWriteQueueMaxSize(int maxSize);
+
+  @Fluent
+  @Override
+  PgCopyIn drainHandler(Handler<Void> handler);
+
+  /**
+   * Completion of the COPY command.
+   * <p>
+   * The future succeeds with the row count reported by PostgreSQL, once the server is ready for
+   * the next command, so the connection can be used again as soon as it is notified.
+   */
+  Future<Integer> completion();
+
+  /**
+   * Fail COPY with a CopyFail message (server will respond with ErrorResponse).
+   */
+  Future<Void> abort(String message);
+}

vertx-pg-client/src/main/java/io/vertx/pgclient/PgCopyInOptions.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2011-2026 Contributors to the Eclipse Foundation
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+ * which is available at https://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+ */
+
+package io.vertx.pgclient;
+
+import io.vertx.codegen.annotations.DataObject;
+import io.vertx.codegen.json.annotations.JsonGen;
+import io.vertx.core.json.JsonObject;
+
+/**
+ * Options for {@link PgConnection#copyIn(String, PgCopyInOptions)}.
+ */
+@DataObject
+@JsonGen(publicConverter = false)
+public class PgCopyInOptions {
+
+  /**
+   * The default target payload size, in bytes = {@code 256 KiB}.
+   */
+  public static final int DEFAULT_CHUNK_SIZE = 256 * 1024;
+
+  private int chunkSize = DEFAULT_CHUNK_SIZE;
+
+  public PgCopyInOptions() {
+  }
+
+  public PgCopyInOptions(PgCopyInOptions other) {
+    this.chunkSize = other.chunkSize;
+  }
+
+  public PgCopyInOptions(JsonObject json) {
+    PgCopyInOptionsConverter.fromJson(json, this);
+  }
+
+  /**
+   * @return the target payload size, in bytes
+   */
+  public int getChunkSize() {
+    return chunkSize;
+  }
+
+  /**
+   * Set the target payload size.
+   * <p/>
+   * Writes are combined until they reach this size before being sent, so that many small writes do
+   * not become many small messages. A write larger than this size is sent on its own.
+   *
+   * @param chunkSize the target payload size, in bytes
+   * @return a reference to this, so the API can be used fluently
+   */
+  public PgCopyInOptions setChunkSize(int chunkSize) {
+    if (chunkSize <= 0) {
+      throw new IllegalArgumentException("chunkSize must be > 0");
+    }
+    this.chunkSize = chunkSize;
+    return this;
+  }
+
+  public JsonObject toJson() {
+    JsonObject json = new JsonObject();
+    PgCopyInOptionsConverter.toJson(this, json);
+    return json;
+  }
+}