Merge pull request #15 from oracle/5-serialized-batch-results

Batch.execute() Serializes Statement Execution

Author: jeandelavarene

src/main/java/oracle/r2dbc/impl/OracleBatchImpl.java

@@ -24,11 +24,16 @@
 import java.sql.Connection;
 import java.util.LinkedList;
 import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.BiFunction;
 
 import io.r2dbc.spi.Batch;
 import io.r2dbc.spi.R2dbcException;
 import io.r2dbc.spi.Result;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
 import io.r2dbc.spi.Statement;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
@@ -104,6 +109,18 @@ public Batch add(String sql) {
    * are executed in the order they were added. Calling this method clears all
    * statements that have been added to the current batch.
    * </p><p>
+   * A {@code Result} emitted by the returned {@code Publisher} must be
+   * <a href="OracleStatementImpl.html#fully-consumed-result">
+   *   fully-consumed
+   * </a>
+   * before the next {@code Result} is emitted. This ensures that a command in
+   * the batch can not be executed while the {@code Result} of a previous
+   * command is consumed concurrently. It is a known limitation of the Oracle
+   * R2DBC Driver that concurrent operations on a single {@code Connection}
+   * will result in blocked threads. Deferring {@code Statement} execution
+   * until full consumption of the previous {@code Statement}'s {@code Result}
+   * is necessary in order to avoid blocked threads.
+   * </p><p>
    * If the execution of any statement in the sequence results in a failure,
    * then the returned publisher emits {@code onError} with an
    * {@link R2dbcException} that describes the failure, and all subsequent
@@ -126,17 +143,121 @@ public Publisher<? extends Result> execute() {
     statements = new LinkedList<>();
 
     AtomicBoolean isSubscribed = new AtomicBoolean(false);
-    return Flux.defer(() -> {
-      if (isSubscribed.compareAndSet(false, true)) {
-        return Flux.fromIterable(currentStatements)
-          .concatMap(Statement::execute);
-      }
-      else {
-        return Mono.error(new IllegalStateException(
+    return Flux.defer(() -> isSubscribed.compareAndSet(false, true)
+      ? executeBatch(currentStatements)
+      : Mono.error(new IllegalStateException(
           "Multiple subscribers are not supported by the Oracle R2DBC" +
-            " Batch.execute() publisher"));
-      }
-    });
+            " Batch.execute() publisher")));
+  }
+
+  /**
+   * Executes each {@code Statement} in a {@code Queue} of {@code statements}.
+   * A {@code Statement} is not executed until the {@code Result} of any
+   * previous {@code Statement} is fully-consumed.
+   * @param statements {@code Statement}s to execute. Not null.
+   * @return A {@code Publisher} of each {@code Statement}'s {@code Result}.
+   * Not null.
+   */
+  private static Publisher<? extends Result> executeBatch(
+    Queue<Statement> statements) {
+
+    // Reference a Publisher that terminates when the previous Statement's
+    // Result has been consumed.
+    AtomicReference<Publisher<Void>> previous =
+      new AtomicReference<>(Mono.empty());
+
+    return Flux.fromIterable(statements)
+      .concatMap(statement -> {
+
+        // Complete when this statement's result is consumed
+        CompletableFuture<Void> next = new CompletableFuture<>();
+
+        return Flux.from(statement.execute())
+          // Delay execution by delaying Publisher.subscribe(Subscriber) until the
+          // previous statement's result is consumed.
+          .delaySubscription(
+            // Update the reference; This statement is now the "previous"
+            // statement.
+            previous.getAndSet(Mono.fromCompletionStage(next)))
+          // Batch result completes the "next" future when fully consumed.
+          .map(result -> new BatchResult(next, result));
+      });
+  }
+
+  /**
+   * <p>
+   * A {@code Result} that completes a {@link CompletableFuture} when it has
+   * been fully consumed. Instances of {@code BatchResult} are used by Oracle
+   * R2DBC to ensure that statement execution and row data processing do
+   * not occur concurrently; The completion of the future signals that the row
+   * data of a result has been fully consumed, and that no more database
+   * calls will be initiated to fetch additional rows.
+   * </p><p>
+   * Instances of {@code BatchResult} delegate invocations of
+   * {@link #getRowsUpdated()} and {@link #map(BiFunction)} to a
+   * {@code Result} provided on construction; The behavior of {@code Publisher}s
+   * returned by these methods is identical to those returned by the delegate
+   * {@code Result}.
+   * </p>
+   */
+  private static final class BatchResult implements Result {
+
+    /** Completed when this {@code BatchResult} is fully consumed */
+     final CompletableFuture<Void> consumeFuture;
+
+    /** Delegate {@code Result} that provides row data or an update count */
+    final Result delegateResult;
+
+    /**
+     * Constructs a new result that completes a {@code consumeFuture} when the
+     * row data or update count of a {@code delegateResult} has been fully
+     * consumed.
+     * @param consumeFuture Future completed upon consumption
+     * @param delegateResult Result of row data or an update count
+     */
+    BatchResult(CompletableFuture<Void> consumeFuture, Result delegateResult) {
+      this.consumeFuture = consumeFuture;
+      this.delegateResult = delegateResult;
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Immediately completes the {@link #consumeFuture} and then returns the
+     * update count {@code Publisher} of the {@link #delegateResult}. After
+     * returning an update count {@code Publisher}, the {@link #delegateResult}
+     * can not initiate any more database calls (based on the assumption
+     * noted below).
+     * </p>
+     * @implNote It is assumed that the {@link #delegateResult} will throw
+     * {@link IllegalStateException} upon multiple attempts to consume it, and
+     * this method does not check for multiple consumptions.
+     */
+    @Override
+    public Publisher<Integer> getRowsUpdated() {
+      consumeFuture.complete(null);
+      return Flux.from(delegateResult.getRowsUpdated());
+    }
+
+    /**
+     * {@inheritDoc}
+     * <p>
+     * Completes the {@link #consumeFuture} after the row data {@code
+     * Publisher} of the {@link #delegateResult} emits a terminal signal or
+     * has it's {@code Subscription} cancelled. After emitting a terminal
+     * signal or having it's {@code Subscription} cancelled, the
+     * {@link #delegateResult} can not initiate any more database calls.
+     * </p>
+     * @implNote It is assumed that the {@link #delegateResult} will throw
+     * {@link IllegalStateException} upon multiple attempts to consume it, and
+     * this method does not check for multiple consumptions.
+     */
+    @Override
+    public <T> Publisher<T> map(
+      BiFunction<Row, RowMetadata, ? extends T> mappingFunction) {
+      return Flux.<T>from(delegateResult.map(mappingFunction))
+        .doFinally(signalType -> consumeFuture.complete(null));
+    }
   }
 }
 

src/main/java/oracle/r2dbc/impl/OracleStatementImpl.java

@@ -63,14 +63,14 @@
  * Database and an ORA-01000 error will be raised. The Oracle R2DBC Driver
  * will close cursors after each {@link Result} emitted by the
  * {@link #execute()} publisher has been fully consumed.
- * </p><p>
+ * </p><p id="fully-consumed-result">
  * To ensure that cursors are eventually closed, application code MUST
- * fully consume the {@link Result} objects emitted by the {@link #execute()}
- * publisher. A {@code Result} is fully consumed by subscribing to it's
+ * fully consume {@link Result} objects emitted by the {@link #execute()}
+ * {@code Publisher}. A {@code Result} is fully consumed by subscribing to its
  * {@linkplain Result#getRowsUpdated() update count} or
  * {@linkplain Result#map(BiFunction) row data} {@code Publisher} and then
- * requesting items until the publisher emits {@code onComplete/onError} or
- * cancelling the subscription.
+ * requesting items until the {@code Publisher} emits {@code onComplete/onError}
+ * or its {@code Subscription} is cancelled.
  * </p><p>
  * To improve performance when the same SQL statement is executed multiple
  * times, implementations of {@link ReactiveJdbcAdapter} are expected to
@@ -995,4 +995,4 @@ private interface DeferredBind {
     Mono<Void> discard();
   }
 
-}
+}

src/test/java/oracle/r2dbc/impl/OracleResultImplTest.java

@@ -30,7 +30,10 @@
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Signal;
 
+import java.util.Iterator;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.function.BiFunction;
 
 import static java.util.Arrays.asList;
@@ -65,12 +68,14 @@ public void testGetRowsUpdated() {
         "CREATE TABLE testGetRowsUpdated (x NUMBER, y NUMBER)"));
       try {
         // Expect update count of 1 from each INSERT.
-        List<? extends Result> insertResults =
-          awaitMany(connection.createBatch()
+        Iterator<? extends Result> insertResults =
+          Flux.from(connection.createBatch()
             .add("INSERT INTO testGetRowsUpdated (x, y) VALUES (0, 0)")
             .add("INSERT INTO testGetRowsUpdated (x, y) VALUES (0, 1)")
-            .execute());
-        Result insertResult0 = insertResults.get(0);
+            .execute())
+            .toIterable()
+            .iterator();
+        Result insertResult0 = insertResults.next();
         Publisher<Integer> insertCountPublisher0 =
           insertResult0.getRowsUpdated();
         awaitOne(1, insertCountPublisher0);
@@ -84,7 +89,7 @@ public void testGetRowsUpdated() {
         // Expect update count publisher to support multiple subscribers
         awaitOne(1, insertCountPublisher0);
 
-        Result insertResult1 = insertResults.get(1);
+        Result insertResult1 = insertResults.next();
         Publisher<Integer> insertCountPublisher1 =
           insertResult1.getRowsUpdated();
         awaitOne(1, insertCountPublisher1);
@@ -185,12 +190,14 @@ public void testMap() {
         "CREATE TABLE testMap (x NUMBER, y NUMBER)"));
       try {
         // Expect no row data from each INSERT.
-        List<? extends Result> insertResults =
-          awaitMany(connection.createBatch()
+        Iterator<? extends Result> insertResults =
+          Flux.from(connection.createBatch()
             .add("INSERT INTO testMap (x, y) VALUES (0, 0)")
             .add("INSERT INTO testMap (x, y) VALUES (0, 1)")
-            .execute());
-        Result insertResult0 = insertResults.get(0);
+            .execute())
+            .toIterable()
+            .iterator();
+        Result insertResult0 = insertResults.next();
         Publisher<Object> insertRowPublisher0 =
           insertResult0.map((row, metadata) -> row.get(0));
         awaitNone(insertRowPublisher0);
@@ -204,7 +211,7 @@ public void testMap() {
         // Expect row data publisher to reject multiple subscribers
         awaitError(IllegalStateException.class, insertRowPublisher0);
 
-        Result insertResult1 = insertResults.get(1);
+        Result insertResult1 = insertResults.next();
         Publisher<Object> insertRowPublisher1 =
           insertResult1.map((row, metadata) -> row.get(0));
         awaitNone(insertRowPublisher1);