Batch Support for PostgreSQL

[plombardi89]

Any chance this will be implemented soon or if there is a work around beyond writing against a raw JDBC client? Currently throws an UnsupportedOperationException

2016-11-05 05:09:22.163 ERROR [vert.x-eventloop-thread-2] i.d.m.v.d.PostgresInteractionsManager - Failed to write to database
java.lang.UnsupportedOperationException: Not implemented
    at io.vertx.ext.asyncsql.impl.AsyncSQLConnectionImpl.batchWithParams(AsyncSQLConnectionImpl.java:208)

Sadness.

[pmlopes]

I'm not sure the underlying driver support batch, at least at the API level. An alternative is to implement batching manually by concatenating all queries in a single string.

[wangzhenhui1991]

I saw the API,Batch operations,that supported this operation,but it's not implemented.

  @Override
  public SQLConnection batchWithParams(String sqlStatement, List<JsonArray> args, Handler<AsyncResult<List<Integer>>> handler) {
    // This should be simple in postgres, since it is just append the query separator after each query and send as a big
    // sql statement, however it does not seem to work on mysql
    throw new UnsupportedOperationException("Not implemented");
  }

Do you have any suggestion about this following problem?

                    List<JsonArray> params= new ArrayList<>();
                    String sql= "insert into table values(?,?)";
                    objects.forEach(obj->{
                        params.add(obj);
                    });
                   connection.batchWithParams(sql,params,res->{
                        if(res.failed()){
                            conn.result().close();
                            throw new RuntimeException(res.cause());
                        }else{
                            LOGGER.info(res.result().toString());
                            conn.result().close();
                        }
                    });

Waiting for your reply.

[Narigo]

@wangzhenhui1991 batched execution is currently not supported by the underlying driver :(

[shaykh-salman]

Is there any batch support for PostgreSql with java?

[vietj]

you can look at https://github.com/vietj/reactive-pg-client which supports batching

[sherl0cks]

In the meantime, can we please can some logging or a runtime error thrown from the vert.x client? It's very confusing for a new user when this just hangs. Perhaps getting the docs updated as well to indicate this is a support gap - https://vertx.io/docs/vertx-sql-common/java/ https://vertx.io/docs/vertx-mysql-postgresql-client/java/.

[Sxubas]

@sherl0cks OMG thank you very very much! I was going nuts because I did not know what was going on. I will try the 'concat every sql into a big batch' but how do I add the params to each query in order to prevent injection?

Just in case, as of today, is there any alternative/fix?

EDIT: turns out that I realized what I was asking for was the missing implementation (kindof). I'll take https://github.com/vietj/reactive-pg-client and cry while I refactor the REST I had done with the common SQL interface. It would be great to update the docs to point that batch operations are unsupported.

[sherl0cks]

@vietj what do we need to do get some attention on these docs. This is a nasty paper cut.

[vietj]

@sherl0cks a PR for docs is welcome, it would be back-ported to 3.5 branch and the web-site updated accordingly.

[sherl0cks]

@vietj OK - I was hoping for a different answer, but I'll try to add this to my queue in the coming weeks. An initial review is that we'll need updates in both https://github.com/vert-x3/vertx-mysql-postgresql-client and https://github.com/vert-x3/vertx-sql-common

[vietj]

why is it needed to update vertx-sql-common ?

[melaraj2]

Here is a solution that works for me: You need two methods. The user only calls the public method to use. similar to using the unimplemented batch method.

public static Future<List<UpdateResult>> executeBatch(SQLConnection con, String sql, List<JsonArray> params){
    try {

      Promise<List<UpdateResult>> promise = Promise.promise();
      ConcurrentLinkedQueue<SQLUpdates> statements= new ConcurrentLinkedQueue();

      for (JsonArray param : params) {
        statements.add(new SQLUpdates(sql,param));
      }

      List<UpdateResult> results=new ArrayList<>();

      executeBatch(con,statements,results)
        .onSuccess(res->promise.complete(results))
        .onFailure(promise::fail);
      return promise.future();
    } catch (Throwable t) {
      return Future.failedFuture(t);
    }

  }

  private static Future<Void> executeBatch(SQLConnection con, ConcurrentLinkedQueue<SQLUpdates> statements, List<UpdateResult> results) {
    try {

      var statement= statements.poll();
      if (statement==null)
      {
        return Future.succeededFuture(null);
      }

      Promise<Void> promise = Promise.promise();

      con.updateWithParams(statement.sql,statement.params,res->{
        if (res.succeeded())
        {
          executeBatch(con,statements,results).onSuccess(unused->promise.complete());
          results.add(res.result());

        }else
        {
          promise.fail(res.cause());
        }
      });
      return promise.future();
    } catch (Throwable t) {
      return Future.failedFuture(t);
    }
  } 

  static class SQLUpdates {
    String sql;
    JsonArray params;

    public SQLUpdates(String sql, JsonArray params) {
      this.sql = sql;
      this.params = params;
    }
  }

`