Improve the implementation of prepared statement.

Motivation:

The implementation of prepared statement tries to accomodate both a lazy mode and a non lazy one, leading to the utilization of a future to unify the obtention of the prepared statement.

This is clearly suboptimal for the common case, since lazy prepared statement is only necessary when meeting indeterminate prepared statement (ambiguous) that requires a tuple for an actual preparation.

In addition the execution is done using a context comparison which fails when on a duplicated context.

Changes:

Use the context in thread comparison instead of context equality.

Make a prepared statement base which has two implementations lazy/direct which are both optimized for their specific cases.

Author: vietj

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/CursorImpl.java

@@ -36,7 +36,7 @@
 public class CursorImpl implements Cursor {
 
   private final Connection conn;
-  private final PreparedStatementImpl ps;
+  private final PreparedStatementBase ps;
   private final ContextInternal context;
   private final boolean autoCommit;
   private final TupleInternal params;
@@ -45,7 +45,7 @@ public class CursorImpl implements Cursor {
   private boolean closed;
   private QueryResultBuilder<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> result;
 
-  CursorImpl(PreparedStatementImpl ps, Connection conn, ContextInternal context, boolean autoCommit, TupleInternal params) {
+  CursorImpl(PreparedStatementBase ps, Connection conn, ContextInternal context, boolean autoCommit, TupleInternal params) {
     this.ps = ps;
     this.conn = conn;
     this.context = context;

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PreparedStatementImpl.java renamed to vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/PreparedStatementBase.java

@@ -44,45 +44,92 @@
 /**
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
-public class PreparedStatementImpl implements PreparedStatement {
+public abstract class PreparedStatementBase implements PreparedStatement {
 
   public static PreparedStatement create(Connection conn, ContextInternal context, io.vertx.sqlclient.internal.PreparedStatement ps, boolean autoCommit) {
-    return new PreparedStatementImpl(conn, context, ps, autoCommit);
+    return new PreparedStatementBase(conn, context, ps, autoCommit) {
+      @Override
+      protected void withPreparedStatement2(PrepareOptions options, Tuple args, Handler<AsyncResult<io.vertx.sqlclient.internal.PreparedStatement>> handler) {
+        handler.handle(Future.succeededFuture(ps));
+      }
+      @Override
+      protected void doClose(Promise<Void> promise) {
+        CloseStatementCommand cmd = new CloseStatementCommand(ps);
+        conn.schedule(cmd, promise);
+      }
+      @Override
+      protected void closeCursor(String cursorId, Promise<Void> promise) {
+        CloseCursorCommand cmd = new CloseCursorCommand(cursorId, ps);
+        conn.schedule(cmd, promise);
+      }
+    };
   }
 
   public static PreparedStatement create(Connection conn, ContextInternal context, PrepareOptions options, String sql, boolean autoCommit) {
-    return new PreparedStatementImpl(conn, context, sql, options, autoCommit);
+    return new PreparedStatementBase(conn, context, sql, options, autoCommit) {
+      Future<io.vertx.sqlclient.internal.PreparedStatement> future;
+      @Override
+      protected void withPreparedStatement2(PrepareOptions options, Tuple args, Handler<AsyncResult<io.vertx.sqlclient.internal.PreparedStatement>> handler) {
+        if (future == null) {
+          Promise<io.vertx.sqlclient.internal.PreparedStatement> promise = context.promise();
+          PrepareStatementCommand prepare = new PrepareStatementCommand(sql, options, true, args.types());
+          conn.schedule(prepare, promise);
+          future = promise.future();
+        }
+        future.onComplete(handler);
+      }
+      @Override
+      protected void doClose(Promise<Void> promise) {
+        if (future != null) {
+          future.onComplete(ar -> {
+            if (ar.succeeded()) {
+              CloseStatementCommand cmd = new CloseStatementCommand(ar.result());
+              conn.schedule(cmd, promise);
+            } else {
+              promise.fail(ar.cause());
+            }
+          });
+        }
+      }
+      @Override
+      protected void closeCursor(String cursorId, Promise<Void> promise) {
+        if (future != null) {
+          future.onComplete(ar -> {
+            if (ar.succeeded()) {
+              CloseCursorCommand cmd = new CloseCursorCommand(cursorId, ar.result());
+              conn.schedule(cmd, promise);
+            } else {
+              promise.fail(ar.cause());
+            }
+          });
+        } else {
+          promise.fail("Invalid");
+        }
+      }
+    };
   }
 
   private final Connection conn;
   private final ContextInternal context;
-  private final String sql;
   private final PrepareOptions options;
-  private Promise<io.vertx.sqlclient.internal.PreparedStatement> promise;
-  private Future<io.vertx.sqlclient.internal.PreparedStatement> future;
   private final boolean autoCommit;
   private final AtomicBoolean closed = new AtomicBoolean();
 
-  private PreparedStatementImpl(Connection conn, ContextInternal context, io.vertx.sqlclient.internal.PreparedStatement ps, boolean autoCommit) {
+  private PreparedStatementBase(Connection conn, ContextInternal context, io.vertx.sqlclient.internal.PreparedStatement ps, boolean autoCommit) {
     this.conn = conn;
     this.context = context;
-    this.sql = null;
     this.options = null;
-    this.promise = null;
-    this.future = Future.succeededFuture(ps);
     this.autoCommit = autoCommit;
   }
 
-  private PreparedStatementImpl(Connection conn,
+  private PreparedStatementBase(Connection conn,
                                 ContextInternal context,
                                 String sql,
                                 PrepareOptions options,
                                 boolean autoCommit) {
     this.conn = conn;
     this.context = context;
-    this.sql = sql;
     this.options = options;
-    this.promise = Promise.promise();
     this.autoCommit = autoCommit;
   }
 
@@ -97,19 +144,15 @@ public PreparedQuery<RowSet<Row>> query() {
   }
 
   void withPreparedStatement(PrepareOptions options, Tuple args, Handler<AsyncResult<io.vertx.sqlclient.internal.PreparedStatement>> handler) {
-    if (context == Vertx.currentContext()) {
-      if (future == null) {
-        // Lazy statement;
-        PrepareStatementCommand prepare = new PrepareStatementCommand(sql, options, true, args.types());
-        conn.schedule(prepare, promise);
-        future = promise.future();
-      }
-      future.onComplete(handler);
+    if (context.inThread()) {
+      withPreparedStatement2(options, args, handler);
     } else {
       context.runOnContext(v -> withPreparedStatement(options, args, handler));
     }
   }
 
+  protected abstract void withPreparedStatement2(PrepareOptions options, Tuple args, Handler<AsyncResult<io.vertx.sqlclient.internal.PreparedStatement>> handler);
+
   <R, F extends SqlResult<R>> void execute(Tuple args,
                                            int fetch,
                                            String cursorId,
@@ -155,27 +198,13 @@ private Cursor cursor(TupleInternal args) {
     return new CursorImpl(this, conn, context, autoCommit, args);
   }
 
+  protected abstract void doClose(Promise<Void> promise);
+
   @Override
   public Future<Void> close() {
     if (closed.compareAndSet(false, true)) {
       Promise<Void> promise = context.promise();
-      if (this.promise == null) {
-        CloseStatementCommand cmd = new CloseStatementCommand(future.result());
-        conn.schedule(cmd, promise);
-      } else {
-        if (future == null) {
-          future = this.promise.future();
-          this.promise.fail("Closed");
-        }
-        future.onComplete(ar -> {
-          if (ar.succeeded()) {
-            CloseStatementCommand cmd = new CloseStatementCommand(ar.result());
-            conn.schedule(cmd, promise);
-          } else {
-            promise.complete();
-          }
-        });
-      }
+      doClose(promise);
       return promise.future();
     } else {
       return context.failedFuture("Already closed");
@@ -187,16 +216,7 @@ public RowStream<Row> createStream(int fetch, Tuple args) {
     return new RowStreamImpl(this, context, fetch, args);
   }
 
-  void closeCursor(String cursorId, Promise<Void> promise) {
-    future.onComplete(ar -> {
-      if (ar.succeeded()) {
-        CloseCursorCommand cmd = new CloseCursorCommand(cursorId, ar.result());
-        conn.schedule(cmd, promise);
-      } else {
-        promise.fail(ar.cause());
-      }
-    });
-  }
+  protected abstract void closeCursor(String cursorId, Promise<Void> promise);
 
   private class PreparedStatementQuery<T, R extends SqlResult<T>> extends QueryBase<T, R> implements PreparedQuery<R> {
 
@@ -232,7 +252,7 @@ public Future<R> execute(Tuple args) {
     }
 
     private void execute(Tuple args, PromiseInternal<R> promise) {
-      PreparedStatementImpl.this.execute(args, 0, null, false, builder, promise);
+      PreparedStatementBase.this.execute(args, 0, null, false, builder, promise);
     }
 
     @Override
@@ -246,7 +266,7 @@ private void executeBatch(List<Tuple> argsList, PromiseInternal<R> promise) {
       if (argsList.isEmpty()) {
         promise.fail("Empty batch");
       } else {
-        PreparedStatementImpl.this.executeBatch(argsList, builder, promise);
+        PreparedStatementBase.this.executeBatch(argsList, builder, promise);
       }
     }
   }

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/RowStreamImpl.java

@@ -32,7 +32,7 @@
 
 public class RowStreamImpl implements RowStreamInternal, Handler<AsyncResult<RowSet<Row>>> {
 
-  private final PreparedStatementImpl ps;
+  private final PreparedStatementBase ps;
   private final ContextInternal context;
   private final int fetch;
   private final Tuple params;
@@ -46,7 +46,7 @@ public class RowStreamImpl implements RowStreamInternal, Handler<AsyncResult<Row
   private boolean readInProgress;
   private Iterator<Row> result;
 
-  RowStreamImpl(PreparedStatementImpl ps, ContextInternal context, int fetch, Tuple params) {
+  RowStreamImpl(PreparedStatementBase ps, ContextInternal context, int fetch, Tuple params) {
     this.ps = ps;
     this.context = context;
     this.fetch = fetch;

vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java

@@ -26,7 +26,7 @@
 import io.vertx.sqlclient.PreparedStatement;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.Transaction;
-import io.vertx.sqlclient.impl.PreparedStatementImpl;
+import io.vertx.sqlclient.impl.PreparedStatementBase;
 import io.vertx.sqlclient.impl.TransactionImpl;
 import io.vertx.sqlclient.internal.command.CommandBase;
 import io.vertx.sqlclient.internal.command.PrepareStatementCommand;
@@ -78,10 +78,10 @@ public Future<PreparedStatement> prepare(String sql, PrepareOptions options) {
     schedule(new PrepareStatementCommand(sql, options, true), promise);
     return promise.future()
       .compose(
-      cr -> Future.succeededFuture(PreparedStatementImpl.create(conn, context, cr, autoCommit())),
+      cr -> Future.succeededFuture(PreparedStatementBase.create(conn, context, cr, autoCommit())),
       err -> {
         if (conn.isIndeterminatePreparedStatementError(err)) {
-          return Future.succeededFuture(PreparedStatementImpl.create(conn, context, options, sql, autoCommit()));
+          return Future.succeededFuture(PreparedStatementBase.create(conn, context, options, sql, autoCommit()));
         } else {
           return Future.failedFuture(err);
         }