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,16 +36,16 @@
 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;
 
   private String id;
   private boolean closed;
-  private QueryResultBuilder<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> result;
+  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;
@@ -64,22 +64,14 @@ public synchronized boolean hasMore() {
   @Override
   public synchronized Future<RowSet<Row>> read(int count) {
     PromiseInternal<RowSet<Row>> promise = context.promise();
-    ps.withPreparedStatement(ps.options(), params, ar -> {
-      if (ar.succeeded()) {
-        PreparedStatement preparedStatement = ar.result();
-        QueryExecutor<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> builder = new QueryExecutor<>(RowSetImpl.FACTORY, RowSetImpl.COLLECTOR);
-        if (id == null) {
-          id = UUID.randomUUID().toString();
-          this.result = builder.executeExtendedQuery(conn, preparedStatement, ps.options(), autoCommit, params, count, id, false, promise);
-        } else if (this.result.isSuspended()) {
-          this.result = builder.executeExtendedQuery(conn, preparedStatement, ps.options(), autoCommit, params, count, id, true, promise);
-        } else {
-          throw new IllegalStateException();
-        }
-      } else {
-        promise.fail(ar.cause());
-      }
-    });
+    boolean suspended;
+    if (id == null) {
+      id = UUID.randomUUID().toString();
+      suspended = false;
+    } else {
+      suspended = true;
+    }
+    ps.readCursor(this, id, suspended, params, count, promise);
     return promise.future();
   }
 

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

@@ -0,0 +1,270 @@
+/*
+ * Copyright (C) 2017 Julien Viet
+ *
+ * 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
+ *
+ * http://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.
+ *
+ */
+
+package io.vertx.sqlclient.impl;
+
+import io.vertx.core.internal.ContextInternal;
+import io.vertx.core.internal.PromiseInternal;
+import io.vertx.sqlclient.PrepareOptions;
+import io.vertx.sqlclient.PreparedQuery;
+import io.vertx.sqlclient.internal.ArrayTuple;
+import io.vertx.sqlclient.internal.Connection;
+import io.vertx.sqlclient.internal.command.CloseCursorCommand;
+import io.vertx.sqlclient.internal.command.CloseStatementCommand;
+import io.vertx.sqlclient.Cursor;
+import io.vertx.sqlclient.PreparedStatement;
+import io.vertx.sqlclient.SqlResult;
+import io.vertx.sqlclient.RowSet;
+import io.vertx.sqlclient.RowStream;
+import io.vertx.sqlclient.Row;
+import io.vertx.sqlclient.Tuple;
+import io.vertx.core.*;
+import io.vertx.sqlclient.internal.command.PrepareStatementCommand;
+import io.vertx.sqlclient.internal.TupleInternal;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.stream.Collector;
+
+/**
+ * @author <a href="mailto:[email protected]">Julien Viet</a>
+ */
+public abstract class PreparedStatementBase implements PreparedStatement {
+
+  public static PreparedStatement create(Connection conn, ContextInternal context, io.vertx.sqlclient.internal.PreparedStatement ps, boolean autoCommit) {
+    return new PreparedStatementBase(conn, context, ps, autoCommit) {
+      @Override
+      protected <R, F extends SqlResult<R>> void executeBatch(List<Tuple> argsList, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p) {
+        builder.executeBatchQuery(conn, options, ps, autoCommit, argsList, p);
+      }
+      @Override
+      protected <R, F extends SqlResult<R>> void execute(Tuple args, int fetch, String cursorId, boolean suspended, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p) {
+        builder.executeExtendedQuery(conn, ps, options, autoCommit, args, fetch, cursorId, suspended,  p);
+      }
+      @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);
+      }
+      @Override
+      protected void readCursor(CursorImpl cursor, String id, boolean suspended, TupleInternal params, int count, PromiseInternal<RowSet<Row>> promise) {
+        QueryExecutor<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> builder = new QueryExecutor<>(RowSetImpl.FACTORY, RowSetImpl.COLLECTOR);
+        cursor.result = builder.executeExtendedQuery(conn, ps, options, autoCommit, params, count, id, suspended, promise);
+      }
+    };
+  }
+
+  public static PreparedStatement create(Connection conn, ContextInternal context, PrepareOptions options, String sql, boolean autoCommit) {
+    return new PreparedStatementBase(conn, context, sql, options, autoCommit) {
+      Future<io.vertx.sqlclient.internal.PreparedStatement> future;
+      void withPreparedStatement(PrepareOptions options, Tuple args, Handler<AsyncResult<io.vertx.sqlclient.internal.PreparedStatement>> handler) {
+        if (context.inThread()) {
+          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);
+        } else {
+          context.runOnContext(v -> withPreparedStatement(options, args, handler));
+        }
+      }
+      @Override
+      protected <R, F extends SqlResult<R>> void executeBatch(List<Tuple> argsList, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p) {
+        withPreparedStatement(options, argsList.get(0), ar -> {
+          if (ar.succeeded()) {
+            builder.executeBatchQuery(conn, options, ar.result(), autoCommit, argsList, p);
+          } else {
+            p.fail(ar.cause());
+          }
+        });
+      }
+      @Override
+      protected <R, F extends SqlResult<R>> void execute(Tuple args, int fetch, String cursorId, boolean suspended, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p) {
+        withPreparedStatement(options, args, ar -> {
+          if (ar.succeeded()) {
+            builder.executeExtendedQuery(conn, ar.result(), options, autoCommit, args, fetch, cursorId, suspended,  p);
+          } else {
+            p.fail(ar.cause());
+          }
+        });
+      }
+      @Override
+      protected void readCursor(CursorImpl cursor, String id, boolean suspended, TupleInternal params, int count, PromiseInternal<RowSet<Row>> promise) {
+        withPreparedStatement(options, params, ar -> {
+          if (ar.succeeded()) {
+            QueryExecutor<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> builder = new QueryExecutor<>(RowSetImpl.FACTORY, RowSetImpl.COLLECTOR);
+            cursor.result = builder.executeExtendedQuery(conn, ar.result(), options, autoCommit, params, count, id, suspended, promise);
+          } else {
+            promise.fail(ar.cause());
+          }
+        });
+      }
+      @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;
+  protected final PrepareOptions options;
+  private final boolean autoCommit;
+  private final AtomicBoolean closed = new AtomicBoolean();
+
+  private PreparedStatementBase(Connection conn, ContextInternal context, io.vertx.sqlclient.internal.PreparedStatement ps, boolean autoCommit) {
+    this.conn = conn;
+    this.context = context;
+    this.options = null;
+    this.autoCommit = autoCommit;
+  }
+
+  private PreparedStatementBase(Connection conn,
+                                ContextInternal context,
+                                String sql,
+                                PrepareOptions options,
+                                boolean autoCommit) {
+    this.conn = conn;
+    this.context = context;
+    this.options = options;
+    this.autoCommit = autoCommit;
+  }
+
+  @Override
+  public PreparedQuery<RowSet<Row>> query() {
+    QueryExecutor<RowSet<Row>, RowSetImpl<Row>, RowSet<Row>> builder = new QueryExecutor<>(RowSetImpl.FACTORY, RowSetImpl.COLLECTOR);
+    return new PreparedStatementQuery<>(builder);
+  }
+
+
+  protected abstract <R, F extends SqlResult<R>> void execute(Tuple args, int fetch, String cursorId, boolean suspended, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p);
+  protected abstract <R, F extends SqlResult<R>> void executeBatch(List<Tuple> argsList, QueryExecutor<R, ?, F> builder, PromiseInternal<F> p);
+  protected abstract void doClose(Promise<Void> promise);
+  protected abstract void closeCursor(String cursorId, Promise<Void> promise);
+  protected abstract void readCursor(CursorImpl cursor, String id, boolean suspended, TupleInternal params, int count, PromiseInternal<RowSet<Row>> promise);
+
+  @Override
+  public Cursor cursor(Tuple args) {
+    return cursor((TupleInternal) args);
+  }
+
+  private Cursor cursor(TupleInternal args) {
+    return new CursorImpl(this, conn, context, autoCommit, args);
+  }
+
+  @Override
+  public Future<Void> close() {
+    if (closed.compareAndSet(false, true)) {
+      Promise<Void> promise = context.promise();
+      doClose(promise);
+      return promise.future();
+    } else {
+      return context.failedFuture("Already closed");
+    }
+  }
+
+  @Override
+  public RowStream<Row> createStream(int fetch, Tuple args) {
+    return new RowStreamImpl(this, context, fetch, args);
+  }
+
+  private class PreparedStatementQuery<T, R extends SqlResult<T>> extends QueryBase<T, R> implements PreparedQuery<R> {
+
+    public PreparedStatementQuery(QueryExecutor<T, ?, R> builder) {
+      super(builder);
+    }
+
+    @Override
+    protected <T2, R2 extends SqlResult<T2>> QueryBase<T2, R2> copy(QueryExecutor<T2, ?, R2> builder) {
+      return new PreparedStatementQuery<>(builder);
+    }
+
+    @Override
+    public <U> PreparedQuery<SqlResult<U>> collecting(Collector<Row, ?, U> collector) {
+      return (PreparedQuery<SqlResult<U>>) super.collecting(collector);
+    }
+
+    @Override
+    public <U> PreparedQuery<RowSet<U>> mapping(Function<Row, U> mapper) {
+      return (PreparedQuery<RowSet<U>>) super.mapping(mapper);
+    }
+
+    @Override
+    public Future<R> execute() {
+      return execute(ArrayTuple.EMPTY);
+    }
+
+    @Override
+    public Future<R> execute(Tuple args) {
+      PromiseInternal<R> promise = context.promise();
+      execute(args, promise);
+      return promise.future();
+    }
+
+    private void execute(Tuple args, PromiseInternal<R> promise) {
+      PreparedStatementBase.this.execute(args, 0, null, false, builder, promise);
+    }
+
+    @Override
+    public Future<R> executeBatch(List<Tuple> argsList) {
+      PromiseInternal<R> promise = context.promise();
+      executeBatch(argsList, promise);
+      return promise.future();
+    }
+
+    private void executeBatch(List<Tuple> argsList, PromiseInternal<R> promise) {
+      if (argsList.isEmpty()) {
+        promise.fail("Empty batch");
+      } else {
+        PreparedStatementBase.this.executeBatch(argsList, builder, promise);
+      }
+    }
+  }
+}