Support implicit pre-read in Consensus Commit

core/src/integration-test/java/com/scalar/db/transaction/jdbc/JdbcTransactionIntegrationTest.java

@@ -36,4 +36,8 @@ public void abort_forOngoingTransaction_ShouldAbortCorrectly() {}
   @Disabled("JDBC transaction doesn't support rollback()")
   @Override
   public void rollback_forOngoingTransaction_ShouldRollbackCorrectly() {}
+
+  @Disabled
+  @Override
+  public void putAndCommit_BlindPutGivenForExisting_ShouldThrowCommitConflictException() {}
 }

core/src/main/java/com/scalar/db/api/OperationBuilder.java

@@ -281,6 +281,23 @@ interface ClearValues<T> {
     T clearValue(String columnName);
   }
 
+  interface Blind<T> {
+    /**
+     * Sets this put operation as a blind write.
+     *
+     * @return the operation builder
+     */
+    T blind();
+
+    /**
+     * Sets whether this put operation is a blind write or not.
+     *
+     * @param blind {@code true} if this put operation is a blind write, {@code false} otherwise
+     * @return the operation builder
+     */
+    T blind(boolean blind);
+  }
+
   interface Limit<T> {
     /**
      * Sets the specified number of results to be returned

core/src/main/java/com/scalar/db/api/Put.java

@@ -39,6 +39,8 @@ public class Put extends Mutation {
 
   private final Map<String, Column<?>> columns;
 
+  private boolean blind;
+
   /**
    * Constructs a {@code Put} with the specified partition {@link Key}.
    *
@@ -80,6 +82,7 @@ public Put(Key partitionKey, Key clusteringKey) {
   public Put(Put put) {
     super(put);
     columns = new LinkedHashMap<>(put.columns);
+    blind = put.blind;
   }
 
   /**
@@ -705,6 +708,7 @@ private void checkIfExists(String name) {
       throw new IllegalArgumentException(name + " doesn't exist");
     }
   }
+
   /**
    * @deprecated As of release 3.6.0. Will be removed in release 5.0.0. Use the setter method of the
    *     Put builder instead; to create a Put builder, use {@link Put#newBuilder()}
@@ -750,6 +754,25 @@ public Put withCondition(MutationCondition condition) {
     return (Put) super.withCondition(condition);
   }
 
+  /**
+   * Returns whether this Put is a blind write.
+   *
+   * @return whether this Put is a blind write
+   */
+  public boolean isBlind() {
+    return blind;
+  }
+
+  /**
+   * Sets whether this Put is a blind write.
+   *
+   * @param blind whether this Put is a blind write
+   */
+  Put setBlind(boolean blind) {
+    this.blind = blind;
+    return this;
+  }
+
   /**
    * Indicates whether some other object is "equal to" this object. The other object is considered
    * equal if:
@@ -775,12 +798,12 @@ public boolean equals(Object o) {
       return false;
     }
     Put other = (Put) o;
-    return columns.equals(other.columns);
+    return columns.equals(other.columns) && blind == other.blind;
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(super.hashCode(), columns);
+    return Objects.hash(super.hashCode(), columns, blind);
   }
 
   @Override
@@ -793,6 +816,7 @@ public String toString() {
         .add("columns", getColumns())
         .add("consistency", getConsistency())
         .add("condition", getCondition())
+        .add("blind", isBlind())
         .toString();
   }
 }

core/src/main/java/com/scalar/db/api/PutBuilder.java

@@ -2,6 +2,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import com.scalar.db.api.OperationBuilder.Blind;
 import com.scalar.db.api.OperationBuilder.ClearClusteringKey;
 import com.scalar.db.api.OperationBuilder.ClearCondition;
 import com.scalar.db.api.OperationBuilder.ClearNamespace;
@@ -76,11 +77,13 @@ public static class Buildable extends OperationBuilder.Buildable<Put>
       implements ClusteringKey<Buildable>,
           Consistency<Buildable>,
           Condition<Buildable>,
-          Values<Buildable> {
+          Values<Buildable>,
+          Blind<Buildable> {
     final Map<String, Column<?>> columns = new LinkedHashMap<>();
     @Nullable Key clusteringKey;
     @Nullable com.scalar.db.api.Consistency consistency;
     @Nullable MutationCondition condition;
+    boolean blind;
 
     private Buildable(@Nullable String namespace, String table, Key partitionKey) {
       super(namespace, table, partitionKey);
@@ -199,6 +202,18 @@ public Buildable value(Column<?> column) {
       return this;
     }
 
+    @Override
+    public Buildable blind() {
+      blind = true;
+      return this;
+    }
+
+    @Override
+    public Buildable blind(boolean blind) {
+      this.blind = blind;
+      return this;
+    }
+
     @Override
     public Put build() {
       Put put = new Put(partitionKey, clusteringKey);
@@ -210,6 +225,7 @@ public Put build() {
       if (condition != null) {
         put.withCondition(condition);
       }
+      put.setBlind(blind);
 
       return put;
     }
@@ -237,6 +253,7 @@ public static class BuildableFromExisting extends Buildable
       this.columns.putAll(put.getColumns());
       this.consistency = put.getConsistency();
       this.condition = put.getCondition().orElse(null);
+      this.blind = put.isBlind();
     }
 
     @Override
@@ -391,5 +408,17 @@ public BuildableFromExisting clearNamespace() {
       this.namespaceName = null;
       return this;
     }
+
+    @Override
+    public BuildableFromExisting blind(boolean blind) {
+      super.blind(blind);
+      return this;
+    }
+
+    @Override
+    public BuildableFromExisting blind() {
+      super.blind();
+      return this;
+    }
   }
 }

core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommit.java

@@ -13,7 +13,9 @@
 import com.scalar.db.api.Selection;
 import com.scalar.db.common.AbstractDistributedTransaction;
 import com.scalar.db.exception.storage.ExecutionException;
+import com.scalar.db.exception.transaction.CommitConflictException;
 import com.scalar.db.exception.transaction.CommitException;
+import com.scalar.db.exception.transaction.CrudConflictException;
 import com.scalar.db.exception.transaction.CrudException;
 import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -94,7 +96,7 @@ public void put(Put put) throws CrudException {
 
   @Override
   public void put(List<Put> puts) throws CrudException {
-    checkArgument(puts.size() != 0);
+    checkArgument(!puts.isEmpty());
     for (Put p : puts) {
       put(p);
     }
@@ -109,15 +111,15 @@ public void delete(Delete delete) throws CrudException {
 
   @Override
   public void delete(List<Delete> deletes) throws CrudException {
-    checkArgument(deletes.size() != 0);
+    checkArgument(!deletes.isEmpty());
     for (Delete d : deletes) {
       delete(d);
     }
   }
 
   @Override
   public void mutate(List<? extends Mutation> mutations) throws CrudException {
-    checkArgument(mutations.size() != 0);
+    checkArgument(!mutations.isEmpty());
     for (Mutation m : mutations) {
       if (m instanceof Put) {
         put((Put) m);
@@ -129,6 +131,19 @@ public void mutate(List<? extends Mutation> mutations) throws CrudException {
 
   @Override
   public void commit() throws CommitException, UnknownTransactionStatusException {
+    // Fill the read set with records from the write and delete sets if they are unread
+    try {
+      crud.fillReadSetForRecordsFromWriteAndDeleteSetsIfUnread();
+    } catch (CrudConflictException e) {
+      throw new CommitConflictException(
+          "Conflict occurred while reading unread records in the write and delete sets",
+          e,
+          getId());
+    } catch (CrudException e) {
+      throw new CommitException(
+          "Failed to read unread records in the write and delete sets", e, getId());
+    }
+
     commit.commit(crud.getSnapshot());
   }
 

core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitConfig.java

@@ -30,6 +30,9 @@ public class ConsensusCommitConfig {
   public static final String ASYNC_COMMIT_ENABLED = PREFIX + "async_commit.enabled";
   public static final String ASYNC_ROLLBACK_ENABLED = PREFIX + "async_rollback.enabled";
 
+  public static final String PARALLEL_FILL_READ_SET_ENABLED =
+      PREFIX + "parallel_fill_read_set.enabled";
+
   public static final int DEFAULT_PARALLEL_EXECUTOR_COUNT = 128;
 
   public static final String INCLUDE_METADATA_ENABLED = PREFIX + "include_metadata.enabled";
@@ -48,6 +51,8 @@ public class ConsensusCommitConfig {
 
   private final boolean isIncludeMetadataEnabled;
 
+  private final boolean parallelFillReadSetEnabled;
+
   public ConsensusCommitConfig(DatabaseConfig databaseConfig) {
     String transactionManager = databaseConfig.getTransactionManager();
     if (!"consensus-commit".equals(transactionManager)) {
@@ -102,10 +107,16 @@ public ConsensusCommitConfig(DatabaseConfig databaseConfig) {
             databaseConfig.getProperties(), PARALLEL_ROLLBACK_ENABLED, parallelCommitEnabled);
 
     asyncCommitEnabled = getBoolean(databaseConfig.getProperties(), ASYNC_COMMIT_ENABLED, false);
+
+    // Use the value of async commit for async rollback as default value
     asyncRollbackEnabled =
         getBoolean(databaseConfig.getProperties(), ASYNC_ROLLBACK_ENABLED, asyncCommitEnabled);
+
     isIncludeMetadataEnabled =
         getBoolean(databaseConfig.getProperties(), INCLUDE_METADATA_ENABLED, false);
+
+    parallelFillReadSetEnabled =
+        getBoolean(databaseConfig.getProperties(), PARALLEL_FILL_READ_SET_ENABLED, true);
   }
 
   public Isolation getIsolation() {
@@ -151,4 +162,8 @@ public boolean isAsyncRollbackEnabled() {
   public boolean isIncludeMetadataEnabled() {
     return isIncludeMetadataEnabled;
   }
+
+  public boolean isParallelFillReadSetEnabled() {
+    return parallelFillReadSetEnabled;
+  }
 }

core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java

@@ -184,7 +184,8 @@ DistributedTransaction begin(String txId, Isolation isolation, SerializableStrat
     Snapshot snapshot =
         new Snapshot(txId, isolation, strategy, tableMetadataManager, parallelExecutor);
     CrudHandler crud =
-        new CrudHandler(storage, snapshot, tableMetadataManager, isIncludeMetadataEnabled);
+        new CrudHandler(
+            storage, snapshot, tableMetadataManager, isIncludeMetadataEnabled, parallelExecutor);
     ConsensusCommit consensus =
         new ConsensusCommit(crud, commit, recovery, mutationOperationChecker);
     getNamespace().ifPresent(consensus::withNamespace);