Remove EXTRA_WRITE strategy in Consensus Commit (#2597)

Author: brfrn169

conf/database.properties

@@ -17,11 +17,6 @@ scalar.db.password=cassandra
 # Isolation level used for Consensus Commit. Either `SNAPSHOT` or `SERIALIZABLE` can be specified. The default is `SNAPSHOT`.
 #scalar.db.consensus_commit.isolation_level=
 
-# Serializable strategy used for Consensus Commit.
-# Either `EXTRA_READ` or `EXTRA_WRITE` can be specified. The default is `EXTRA_READ`.
-# If `SNAPSHOT` is specified in the property `scalar.db.consensus_commit.isolation_level`, this is ignored.
-#scalar.db.consensus_commit.serializable_strategy=
-
 # The given namespace name will be used by operations that do not already specify a namespace.
 # By default, no default namespace name is set on operations.
 #scalar.db.default_namespace_name=<a_namespace_name>

core/src/main/java/com/scalar/db/common/error/CoreError.java

@@ -458,11 +458,11 @@ public enum CoreError implements ScalarDbError {
       ""),
   CONSENSUS_COMMIT_READING_ALREADY_WRITTEN_DATA_NOT_ALLOWED(
       Category.USER_ERROR, "0106", "Reading already-written data is not allowed", "", ""),
-  CONSENSUS_COMMIT_TRANSACTION_NOT_VALIDATED_IN_EXTRA_READ(
+  CONSENSUS_COMMIT_TRANSACTION_NOT_VALIDATED_IN_SERIALIZABLE(
       Category.USER_ERROR,
       "0107",
       "The transaction is not validated."
-          + " When using the EXTRA_READ serializable strategy, you need to call validate()"
+          + " When using the SERIALIZABLE isolation level, you need to call validate()"
           + " before calling commit()",
       "",
       ""),
@@ -929,13 +929,7 @@ public enum CoreError implements ScalarDbError {
       "The condition on the column '%s' is not satisfied",
       "",
       ""),
-  CONSENSUS_COMMIT_READING_EMPTY_RECORDS_IN_EXTRA_WRITE(
-      Category.CONCURRENCY_ERROR,
-      "0021",
-      "Reading empty records might cause a write skew anomaly, so the transaction has been aborted for safety purposes",
-      "",
-      ""),
-  CONSENSUS_COMMIT_ANTI_DEPENDENCY_FOUND_IN_EXTRA_READ(
+  CONSENSUS_COMMIT_ANTI_DEPENDENCY_FOUND(
       Category.CONCURRENCY_ERROR,
       "0022",
       "An anti-dependency was found. The transaction has been aborted",

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

@@ -205,8 +205,8 @@ private void prepareRecords(Snapshot snapshot)
 
   public void validate(Snapshot snapshot) throws ValidationException {
     try {
-      // validation is executed when SERIALIZABLE with EXTRA_READ strategy is chosen.
-      snapshot.toSerializableWithExtraRead(storage);
+      // validation is executed when SERIALIZABLE is chosen.
+      snapshot.toSerializable(storage);
     } catch (ExecutionException e) {
       throw new ValidationException(
           CoreError.CONSENSUS_COMMIT_VALIDATION_FAILED.buildMessage(), e, snapshot.getId());

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

@@ -11,7 +11,6 @@
 import com.scalar.db.api.Consistency;
 import com.scalar.db.api.Delete;
 import com.scalar.db.api.DeleteIf;
-import com.scalar.db.api.Get;
 import com.scalar.db.api.Mutation;
 import com.scalar.db.api.Operation;
 import com.scalar.db.api.Put;
@@ -65,9 +64,8 @@ private void add(Delete base, @Nullable TransactionResult result) throws Executi
 
   private void add(Selection base, @Nullable TransactionResult result) throws ExecutionException {
     if (result == null) {
-      // for deleting non-existing record that was prepared with DELETED for Serializable with
-      // Extra-write
-      mutations.add(composeDelete(base, null));
+      throw new AssertionError(
+          "This path should not be reached since the EXTRA_WRITE strategy is deleted");
     } else if (result.getState().equals(TransactionState.PREPARED)) {
       // for rollforward in lazy recovery
       mutations.add(composePut(base, result));
@@ -122,10 +120,8 @@ private Key getPartitionKey(Operation base, @Nullable TransactionResult result)
         TransactionTableMetadata metadata = tableMetadataManager.getTransactionTableMetadata(base);
         return ScalarDbUtils.getPartitionKey(result, metadata.getTableMetadata());
       } else {
-        // for deleting non-existing record that was prepared with DELETED for Serializable with
-        // Extra-write
-        assert base instanceof Get;
-        return base.getPartitionKey();
+        throw new AssertionError(
+            "This path should not be reached since the EXTRA_WRITE strategy is deleted");
       }
     }
   }
@@ -142,10 +138,8 @@ private Optional<Key> getClusteringKey(Operation base, @Nullable TransactionResu
         TransactionTableMetadata metadata = tableMetadataManager.getTransactionTableMetadata(base);
         return ScalarDbUtils.getClusteringKey(result, metadata.getTableMetadata());
       } else {
-        // for deleting non-existing record that was prepared with DELETED for Serializable with
-        // Extra-write
-        assert base instanceof Get;
-        return base.getClusteringKey();
+        throw new AssertionError(
+            "This path should not be reached since the EXTRA_WRITE strategy is deleted");
       }
     }
   }

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

@@ -22,7 +22,6 @@ public class ConsensusCommitConfig {
   public static final String TRANSACTION_MANAGER_NAME = "consensus-commit";
   public static final String PREFIX = DatabaseConfig.PREFIX + "consensus_commit.";
   public static final String ISOLATION_LEVEL = PREFIX + "isolation_level";
-  public static final String SERIALIZABLE_STRATEGY = PREFIX + "serializable_strategy";
   public static final String COORDINATOR_NAMESPACE = PREFIX + "coordinator.namespace";
 
   public static final String PARALLEL_EXECUTOR_COUNT = PREFIX + "parallel_executor_count";
@@ -64,7 +63,6 @@ public class ConsensusCommitConfig {
   public static final int DEFAULT_COORDINATOR_GROUP_COMMIT_TIMEOUT_CHECK_INTERVAL_MILLIS = 20;
 
   private final Isolation isolation;
-  private final SerializableStrategy strategy;
   @Nullable private final String coordinatorNamespace;
 
   private final int parallelExecutorCount;
@@ -113,14 +111,16 @@ public ConsensusCommitConfig(DatabaseConfig databaseConfig) {
                 .toUpperCase(Locale.ROOT));
     if (isolation.equals(Isolation.SERIALIZABLE)) {
       validateCrossPartitionScanConfig(databaseConfig);
+
+      if (databaseConfig
+          .getProperties()
+          .containsKey("scalar.db.consensus_commit.serializable_strategy")) {
+        logger.warn(
+            "The property \"scalar.db.consensus_commit.serializable_strategy\" is deprecated and will "
+                + "be removed in 5.0.0. The EXTRA_READ strategy is always used for the SERIALIZABLE "
+                + "isolation level.");
+      }
     }
-    strategy =
-        SerializableStrategy.valueOf(
-            getString(
-                    databaseConfig.getProperties(),
-                    SERIALIZABLE_STRATEGY,
-                    SerializableStrategy.EXTRA_READ.toString())
-                .toUpperCase(Locale.ROOT));
 
     coordinatorNamespace = getString(databaseConfig.getProperties(), COORDINATOR_NAMESPACE, null);
 
@@ -193,10 +193,6 @@ public Isolation getIsolation() {
     return isolation;
   }
 
-  public SerializableStrategy getSerializableStrategy() {
-    return strategy;
-  }
-
   public Optional<String> getCoordinatorNamespace() {
     return Optional.ofNullable(coordinatorNamespace);
   }

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

@@ -134,49 +134,49 @@ private CommitHandler createCommitHandler() {
 
   @Override
   public DistributedTransaction begin() {
-    return begin(config.getIsolation(), config.getSerializableStrategy());
+    return begin(config.getIsolation());
   }
 
   @Override
   public DistributedTransaction begin(String txId) {
-    return begin(txId, config.getIsolation(), config.getSerializableStrategy());
+    return begin(txId, config.getIsolation());
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @Deprecated
   @Override
   public DistributedTransaction start(com.scalar.db.api.Isolation isolation) {
-    return begin(Isolation.valueOf(isolation.name()), config.getSerializableStrategy());
+    return begin(Isolation.valueOf(isolation.name()));
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @Deprecated
   @Override
   public DistributedTransaction start(String txId, com.scalar.db.api.Isolation isolation) {
-    return begin(txId, Isolation.valueOf(isolation.name()), config.getSerializableStrategy());
+    return begin(txId, Isolation.valueOf(isolation.name()));
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @Deprecated
   @Override
   public DistributedTransaction start(
       com.scalar.db.api.Isolation isolation, com.scalar.db.api.SerializableStrategy strategy) {
-    return begin(Isolation.valueOf(isolation.name()), (SerializableStrategy) strategy);
+    return begin(Isolation.valueOf(isolation.name()));
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @Deprecated
   @Override
   public DistributedTransaction start(com.scalar.db.api.SerializableStrategy strategy) {
-    return begin(Isolation.SERIALIZABLE, (SerializableStrategy) strategy);
+    return begin(Isolation.SERIALIZABLE);
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
   @Deprecated
   @Override
   public DistributedTransaction start(
       String txId, com.scalar.db.api.SerializableStrategy strategy) {
-    return begin(txId, Isolation.SERIALIZABLE, (SerializableStrategy) strategy);
+    return begin(txId, Isolation.SERIALIZABLE);
   }
 
   /** @deprecated As of release 2.4.0. Will be removed in release 4.0.0. */
@@ -186,31 +186,29 @@ public DistributedTransaction start(
       String txId,
       com.scalar.db.api.Isolation isolation,
       com.scalar.db.api.SerializableStrategy strategy) {
-    return begin(txId, Isolation.valueOf(isolation.name()), (SerializableStrategy) strategy);
+    return begin(txId, Isolation.valueOf(isolation.name()));
   }
 
   @VisibleForTesting
-  DistributedTransaction begin(Isolation isolation, SerializableStrategy strategy) {
+  DistributedTransaction begin(Isolation isolation) {
     String txId = UUID.randomUUID().toString();
-    return begin(txId, isolation, strategy);
+    return begin(txId, isolation);
   }
 
   @VisibleForTesting
-  DistributedTransaction begin(String txId, Isolation isolation, SerializableStrategy strategy) {
+  DistributedTransaction begin(String txId, Isolation isolation) {
     checkArgument(!Strings.isNullOrEmpty(txId));
     checkNotNull(isolation);
     if (isGroupCommitEnabled()) {
       assert groupCommitter != null;
       txId = groupCommitter.reserve(txId);
     }
-    if (!config.getIsolation().equals(isolation)
-        || !config.getSerializableStrategy().equals(strategy)) {
+    if (!config.getIsolation().equals(isolation)) {
       logger.warn(
-          "Setting different isolation level or serializable strategy from the ones "
-              + "in DatabaseConfig might cause unexpected anomalies");
+          "Setting different isolation level from the one in DatabaseConfig might cause unexpected "
+              + "anomalies");
     }
-    Snapshot snapshot =
-        new Snapshot(txId, isolation, strategy, tableMetadataManager, parallelExecutor);
+    Snapshot snapshot = new Snapshot(txId, isolation, tableMetadataManager, parallelExecutor);
     CrudHandler crud =
         new CrudHandler(
             storage, snapshot, tableMetadataManager, isIncludeMetadataEnabled, parallelExecutor);

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

@@ -14,12 +14,10 @@
 import com.scalar.db.api.Operation;
 import com.scalar.db.api.Put;
 import com.scalar.db.api.PutBuilder;
-import com.scalar.db.api.PutIfNotExists;
 import com.scalar.db.api.TransactionState;
 import com.scalar.db.exception.storage.ExecutionException;
 import com.scalar.db.io.Column;
 import com.scalar.db.io.IntColumn;
-import com.scalar.db.io.Value;
 import java.util.ArrayList;
 import java.util.List;
 import javax.annotation.Nullable;
@@ -45,7 +43,8 @@ public void add(Operation base, @Nullable TransactionResult result) throws Execu
     } else if (base instanceof Delete) {
       add((Delete) base, result);
     } else if (base instanceof Get) {
-      add((Get) base);
+      throw new AssertionError(
+          "This path should not be reached since the EXTRA_WRITE strategy is deleted");
     } else {
       throw new AssertionError("PrepareMutationComposer.add only accepts Put, Delete, or Get");
     }
@@ -133,28 +132,6 @@ private void add(Delete base, @Nullable TransactionResult result) throws Executi
     mutations.add(putBuilder.build());
   }
 
-  // This prepares a record that was read but didn't exist to avoid anti-dependency for the record.
-  // This is only called when Serializable with Extra-write strategy is enabled.
-  private void add(Get base) {
-    Put put =
-        new Put(base.getPartitionKey(), base.getClusteringKey().orElse(null))
-            .forNamespace(base.forNamespace().get())
-            .forTable(base.forTable().get())
-            .withConsistency(Consistency.LINEARIZABLE);
-
-    List<Value<?>> values = new ArrayList<>();
-    values.add(Attribute.toIdValue(id));
-    values.add(Attribute.toStateValue(TransactionState.DELETED));
-    values.add(Attribute.toPreparedAtValue(current));
-    values.add(Attribute.toVersionValue(1));
-
-    // check if the record is not interrupted by other conflicting transactions
-    put.withCondition(new PutIfNotExists());
-
-    put.withValues(values);
-    mutations.add(put);
-  }
-
   private List<Column<?>> createBeforeColumns(Mutation base, TransactionResult result)
       throws ExecutionException {
     List<Column<?>> columns = new ArrayList<>();

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

@@ -156,11 +156,8 @@ private Optional<TransactionResult> getLatestResult(
         clusteringKey =
             ScalarDbUtils.getClusteringKey(result, metadata.getTableMetadata()).orElse(null);
       } else {
-        // for deleting non-existing record that was prepared with DELETED for Serializable with
-        // Extra-write
-        assert operation instanceof Get;
-        partitionKey = operation.getPartitionKey();
-        clusteringKey = operation.getClusteringKey().orElse(null);
+        throw new AssertionError(
+            "This path should not be reached since the EXTRA_WRITE strategy is deleted");
       }
     }
 

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

@@ -6,8 +6,11 @@
  * A Serializable strategy used in Consensus Commit algorithm. Both strategies basically make
  * transactions avoid an anti-dependency that is the root cause of the anomalies in Snapshot
  * Isolation.
+ *
+ * @deprecated As of release 3.16.0. Will be removed in release 5.0.0
  */
 @SuppressFBWarnings("NM_SAME_SIMPLE_NAME_AS_INTERFACE")
+@Deprecated
 public enum SerializableStrategy implements com.scalar.db.api.SerializableStrategy {
   /**
    * Extra-write strategy converts all read set into write set when preparing records to remove