Add skeleton for COPY feature API

Author: psolomin

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

@@ -733,6 +733,16 @@ public void batchReturning(SqlClient client) {
       });
   }
 
+  public void copyFromStdinReturning(Vertx vertx, PgConnection client) {
+    client.copyFrom(
+      "COPY my_table FROM STDIN (FORMAT csv, HEADER)",
+      vertx.fileSystem().readFile("path/to/file")
+    ).execute().onSuccess(res -> {
+      Long rowsWritten = res.iterator().next().getLong("rowsWritten");
+      System.out.println("rows affected: " + rowsWritten);
+    });
+  }
+
   public void pgBouncer(PgConnectOptions connectOptions) {
     connectOptions.setUseLayer7Proxy(true);
   }

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

@@ -17,15 +17,21 @@
 
 package io.vertx.pgclient;
 
+import io.vertx.core.buffer.Buffer;
 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.Vertx;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.pgclient.impl.PgConnectionImpl;
+import io.vertx.sqlclient.Query;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
 import io.vertx.sqlclient.SqlConnection;
 
+import java.util.List;
+
 /**
  * A connection to Postgres.
  * <P>
@@ -34,6 +40,7 @@
  *   <ul>
  *     <li>Notification</li>
  *     <li>Request Cancellation</li>
+ *     <li>Copy from STDIN / to STDOUT</li>
  *   </ul>
  * </P>
  *
@@ -92,6 +99,40 @@ static Future<PgConnection> connect(Vertx vertx, String connectionUri) {
   @Fluent
   PgConnection noticeHandler(Handler<PgNotice> handler);
 
+  /**
+   * Imports data into a database.
+   *
+   * <p>Use this method when importing opaque bytes, e.g. from a CSV file.
+   *
+   * <p>If you need bulk inserts of POJOs, use {@link io.vertx.sqlclient.PreparedQuery#executeBatch(List)} instead.
+   *
+   * @param sql COPY command (example {@code COPY my_table FROM STDIN (FORMAT csv, HEADER)})
+   * @param from byte stream data will be fetched from
+   * @return result set with single field {@code rowsWritten}
+   */
+  Query<RowSet<Row>> copyFrom(String sql, Future<Buffer> from);
+
+  /**
+   * Exports data from a database with decoding.
+   *
+   * {@code FORMAT} can only be {@code binary}.
+   *
+   * @param sql COPY command (example {@code COPY my_table TO STDOUT (FORMAT binary)})
+   * @return decoded records
+   */
+  Query<RowSet<Row>> copyTo(String sql);
+
+  /**
+   * Exports data from a database as-is, without decoding.
+   *
+   * <p>Use this method when exporting opaque bytes, e.g. to a CSV file.
+   *
+   * @param sql COPY command (example {@code COPY my_table TO STDOUT (FORMAT csv)})
+   * @param to bytes container data will be written to
+   * @return async result
+   */
+  Future<Void> copyTo(String sql, Buffer to);
+
   /**
    * Send a request cancellation message to tell the server to cancel processing request in this connection.
    * <br>Note: Use this with caution because the cancellation signal may or may not have any effect.

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java

@@ -19,6 +19,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
+import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.pgclient.PgConnectOptions;
 import io.vertx.pgclient.PgConnection;
@@ -27,6 +28,9 @@
 import io.vertx.pgclient.impl.codec.NoticeResponse;
 import io.vertx.pgclient.impl.codec.TxFailedEvent;
 import io.vertx.pgclient.spi.PgDriver;
+import io.vertx.sqlclient.Query;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.RowSet;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.Notification;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
@@ -107,6 +111,21 @@ public PgConnection noticeHandler(Handler<PgNotice> handler) {
     return this;
   }
 
+  @Override
+  public Query<RowSet<Row>> copyFrom(String sql, Future<Buffer> from) {
+    return null;
+  }
+
+  @Override
+  public Query<RowSet<Row>> copyTo(String sql) {
+    return null;
+  }
+
+  @Override
+  public Future<Void> copyTo(String sql, Buffer to) {
+    return Future.succeededFuture();
+  }
+
   @Override
   public int processId() {
     return conn.getProcessId();

vertx-pg-client/src/test/java/io/vertx/pgclient/PgConnectionCopyTest.java

@@ -0,0 +1,28 @@
+package io.vertx.pgclient;
+
+import io.vertx.ext.unit.Async;
+import io.vertx.ext.unit.TestContext;
+import org.junit.Test;
+
+public class PgConnectionCopyTest extends PgConnectionTestBase {
+  @Test
+  public void testCopyToRows(TestContext ctx) {
+    Async async = ctx.async();
+    connector.accept(ctx.asyncAssertSuccess(conn -> {
+      deleteFromTestTable(ctx, conn, () -> {
+        insertIntoTestTable(ctx, conn, 10, () -> {
+          PgConnection pgConn = (PgConnection) conn;
+          pgConn.copyTo("COPY my_table TO STDOUT (FORMAT binary)")
+            .execute()
+            .onComplete(ctx.asyncAssertSuccess(result -> {
+                for (int i = 0; i < 2; i++) {
+                  ctx.assertEquals(1, result.rowCount());
+                  result = result.next();
+                }
+                async.complete();
+          }));
+        });
+      });
+    }));
+  }
+}