feat(collation): follow the collation in Consensus Commit snapshot ordering, equality, and key identity

Author: Torch3333

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

@@ -38,6 +38,7 @@
 import com.scalar.db.exception.transaction.RollbackException;
 import com.scalar.db.exception.transaction.TransactionException;
 import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
+import com.scalar.db.io.CollationComparator;
 import com.scalar.db.io.Key;
 import com.scalar.db.service.StorageFactory;
 import com.scalar.db.transaction.consensuscommit.CoordinatorGroupCommitter.CoordinatorGroupCommitKeyManipulator;
@@ -76,6 +77,7 @@ public class ConsensusCommitManager extends AbstractDistributedTransactionManage
   private final ConsensusCommitOperationChecker operationChecker;
   @Nullable private final CoordinatorGroupCommitter groupCommitter;
   private final boolean coordinatorWriteOmissionOnReadOnlyEnabled;
+  private final CollationComparator collationComparator;
 
   @SuppressFBWarnings("EI_EXPOSE_REP2")
   @Inject
@@ -95,14 +97,16 @@ public ConsensusCommitManager(
     groupCommitter = CoordinatorGroupCommitter.from(config).orElse(null);
     coordinatorWriteOmissionOnReadOnlyEnabled =
         config.isCoordinatorWriteOmissionOnReadOnlyEnabled();
+    collationComparator = CollationComparator.from(databaseConfig);
     crud =
         new CrudHandler(
             storage,
             recoveryExecutor,
             tableMetadataManager,
             config.isIncludeMetadataEnabled(),
             config.isIndexEventuallyConsistentReadEnabled(),
-            parallelExecutor);
+            parallelExecutor,
+            collationComparator);
     StorageInfoProvider storageInfoProvider = new StorageInfoProvider(admin);
     commit = createCommitHandler(config, storageInfoProvider);
     isolation = config.getIsolation();
@@ -113,7 +117,8 @@ public ConsensusCommitManager(
             tableMetadataManager,
             virtualTableInfoManager,
             storageInfoProvider,
-            config.isIncludeMetadataEnabled());
+            config.isIncludeMetadataEnabled(),
+            databaseConfig.getCollation());
 
     ConsensusCommitUtils.warnIfBeforeIndexesAreMissing(admin, config);
   }
@@ -135,14 +140,16 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
     groupCommitter = CoordinatorGroupCommitter.from(config).orElse(null);
     coordinatorWriteOmissionOnReadOnlyEnabled =
         config.isCoordinatorWriteOmissionOnReadOnlyEnabled();
+    collationComparator = CollationComparator.from(databaseConfig);
     crud =
         new CrudHandler(
             storage,
             recoveryExecutor,
             tableMetadataManager,
             config.isIncludeMetadataEnabled(),
             config.isIndexEventuallyConsistentReadEnabled(),
-            parallelExecutor);
+            parallelExecutor,
+            collationComparator);
     StorageInfoProvider storageInfoProvider = new StorageInfoProvider(admin);
     commit = createCommitHandler(config, storageInfoProvider);
     isolation = config.getIsolation();
@@ -153,7 +160,8 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
             tableMetadataManager,
             virtualTableInfoManager,
             storageInfoProvider,
-            config.isIncludeMetadataEnabled());
+            config.isIncludeMetadataEnabled(),
+            databaseConfig.getCollation());
 
     ConsensusCommitUtils.warnIfBeforeIndexesAreMissing(admin, config);
   }
@@ -186,6 +194,7 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
     this.groupCommitter = groupCommitter;
     this.coordinatorWriteOmissionOnReadOnlyEnabled =
         config.isCoordinatorWriteOmissionOnReadOnlyEnabled();
+    this.collationComparator = CollationComparator.from(databaseConfig);
     this.isolation = isolation;
     VirtualTableInfoManager virtualTableInfoManager =
         new VirtualTableInfoManager(admin, databaseConfig.getMetadataCacheExpirationTimeSecs());
@@ -195,13 +204,15 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
             tableMetadataManager,
             virtualTableInfoManager,
             storageInfoProvider,
-            config.isIncludeMetadataEnabled());
+            config.isIncludeMetadataEnabled(),
+            databaseConfig.getCollation());
   }
 
   // `groupCommitter` must be set before calling this method.
   private CommitHandler createCommitHandler(
       ConsensusCommitConfig config, StorageInfoProvider storageInfoProvider) {
-    MutationsGrouper mutationsGrouper = new MutationsGrouper(storageInfoProvider);
+    MutationsGrouper mutationsGrouper =
+        new MutationsGrouper(storageInfoProvider, collationComparator);
     if (isGroupCommitEnabled()) {
       return new CommitHandlerWithGroupCommit(
           storage,
@@ -309,7 +320,8 @@ DistributedTransaction begin(
       txId = groupCommitter.reserve(txId);
       groupCommitSlotReserved = true;
     }
-    Snapshot snapshot = new Snapshot(txId, tableMetadataManager, parallelExecutor);
+    Snapshot snapshot =
+        new Snapshot(txId, tableMetadataManager, parallelExecutor, collationComparator);
     TransactionContext context =
         new TransactionContext(
             txId, snapshot, isolation, readOnly, oneOperation, groupCommitSlotReserved);

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

@@ -4,6 +4,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.scalar.db.api.ConditionalExpression;
+import com.scalar.db.api.ConditionalExpression.Operator;
 import com.scalar.db.api.Delete;
 import com.scalar.db.api.DeleteIf;
 import com.scalar.db.api.DeleteIfExists;
@@ -26,6 +27,7 @@
 import com.scalar.db.common.VirtualTableInfoManager;
 import com.scalar.db.common.checker.ConditionChecker;
 import com.scalar.db.exception.storage.ExecutionException;
+import com.scalar.db.io.Collation;
 import com.scalar.db.util.ScalarDbUtils;
 import javax.annotation.concurrent.ThreadSafe;
 
@@ -36,16 +38,41 @@ public class ConsensusCommitOperationChecker {
   private final VirtualTableInfoManager virtualTableInfoManager;
   private final StorageInfoProvider storageInfoProvider;
   private final boolean isIncludeMetadataEnabled;
+  private final Collation collation;
 
   public ConsensusCommitOperationChecker(
       TransactionTableMetadataManager transactionTableMetadataManager,
       VirtualTableInfoManager virtualTableInfoManager,
       StorageInfoProvider storageInfoProvider,
-      boolean isIncludeMetadataEnabled) {
+      boolean isIncludeMetadataEnabled,
+      Collation collation) {
     this.transactionTableMetadataManager = transactionTableMetadataManager;
     this.virtualTableInfoManager = virtualTableInfoManager;
     this.storageInfoProvider = storageInfoProvider;
     this.isIncludeMetadataEnabled = isIncludeMetadataEnabled;
+    this.collation = collation;
+  }
+
+  /**
+   * Rejects a LIKE or NOT_LIKE condition under the ICU collation. This layer re-evaluates a
+   * selection's conditions itself, and that evaluation stays byte-exact for pattern matching at any
+   * collation, so on a backend whose own collation governs LIKE the two disagree. Every isolation
+   * level is affected. See ADR-10 in docs/collation-adr.md.
+   */
+  private void throwIfLikeConditionUnderIcuCollation(Selection selection) {
+    if (collation != Collation.ICU) {
+      return;
+    }
+    for (Selection.Conjunction conjunction : selection.getConjunctions()) {
+      for (ConditionalExpression condition : conjunction.getConditions()) {
+        Operator operator = condition.getOperator();
+        if (operator == Operator.LIKE || operator == Operator.NOT_LIKE) {
+          throw new IllegalArgumentException(
+              CoreError.COLLATION_ICU_LIKE_CONDITION_NOT_SUPPORTED.buildMessage(
+                  operator, selection.forFullTableName().get(), condition.getColumn().getName()));
+        }
+      }
+    }
   }
 
   /**
@@ -58,6 +85,7 @@ public ConsensusCommitOperationChecker(
    */
   public void check(Get get, TransactionContext context) throws ExecutionException {
     throwIfOperationForVirtualTableButNotConsistentVirtualTableReadStorage(get);
+    throwIfLikeConditionUnderIcuCollation(get);
 
     TransactionTableMetadata metadata =
         getTransactionTableMetadata(transactionTableMetadataManager, get);
@@ -115,6 +143,7 @@ public void check(Get get, TransactionContext context) throws ExecutionException
    */
   public void check(Scan scan, TransactionContext context) throws ExecutionException {
     throwIfOperationForVirtualTableButNotConsistentVirtualTableReadStorage(scan);
+    throwIfLikeConditionUnderIcuCollation(scan);
 
     TransactionTableMetadata metadata =
         getTransactionTableMetadata(transactionTableMetadataManager, scan);

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

@@ -34,6 +34,7 @@
 import com.scalar.db.exception.transaction.TransactionNotFoundException;
 import com.scalar.db.exception.transaction.UnsatisfiedConditionException;
 import com.scalar.db.exception.transaction.ValidationException;
+import com.scalar.db.io.CollationComparator;
 import com.scalar.db.io.Key;
 import com.scalar.db.service.StorageFactory;
 import com.scalar.db.util.ScalarDbUtils;
@@ -84,6 +85,7 @@ public class ConsensusCommitParticipant implements TwoPhaseCommitParticipant {
   private final CrudHandler crud;
   private final ParticipantCommitHandler commit;
   private final ConsensusCommitOperationChecker operationChecker;
+  private final CollationComparator collationComparator;
 
   private final ConcurrentMap<String, ParticipantContext> contexts = new ConcurrentHashMap<>();
 
@@ -101,21 +103,23 @@ public ConsensusCommitParticipant(DatabaseConfig databaseConfig) {
     RecoveryHandler recovery = new RecoveryHandler(storage, coordinator, tableMetadataManager);
     this.recoveryExecutor =
         new RecoveryExecutor(storage, coordinator, recovery, tableMetadataManager);
+    this.collationComparator = CollationComparator.from(databaseConfig);
     this.crud =
         new CrudHandler(
             storage,
             recoveryExecutor,
             tableMetadataManager,
             config.isIncludeMetadataEnabled(),
             config.isIndexEventuallyConsistentReadEnabled(),
-            parallelExecutor);
+            parallelExecutor,
+            collationComparator);
     StorageInfoProvider storageInfoProvider = new StorageInfoProvider(admin);
     this.commit =
         new ParticipantCommitHandler(
             storage,
             tableMetadataManager,
             parallelExecutor,
-            new MutationsGrouper(storageInfoProvider),
+            new MutationsGrouper(storageInfoProvider, collationComparator),
             config.isOnePhaseCommitEnabled());
     VirtualTableInfoManager virtualTableInfoManager =
         new VirtualTableInfoManager(admin, databaseConfig.getMetadataCacheExpirationTimeSecs());
@@ -124,7 +128,8 @@ public ConsensusCommitParticipant(DatabaseConfig databaseConfig) {
             tableMetadataManager,
             virtualTableInfoManager,
             storageInfoProvider,
-            config.isIncludeMetadataEnabled());
+            config.isIncludeMetadataEnabled(),
+            databaseConfig.getCollation());
   }
 
   @VisibleForTesting
@@ -135,7 +140,8 @@ public ConsensusCommitParticipant(DatabaseConfig databaseConfig) {
       RecoveryExecutor recoveryExecutor,
       CrudHandler crud,
       ParticipantCommitHandler commit,
-      ConsensusCommitOperationChecker operationChecker) {
+      ConsensusCommitOperationChecker operationChecker,
+      CollationComparator collationComparator) {
     this.config = checkNotNull(config);
     this.participantId = resolveParticipantId(config);
     this.storage = null;
@@ -146,6 +152,7 @@ public ConsensusCommitParticipant(DatabaseConfig databaseConfig) {
     this.crud = checkNotNull(crud);
     this.commit = checkNotNull(commit);
     this.operationChecker = checkNotNull(operationChecker);
+    this.collationComparator = checkNotNull(collationComparator);
   }
 
   private static String resolveParticipantId(ConsensusCommitConfig config) {
@@ -165,7 +172,8 @@ public String getId() {
   @Override
   public void join(String transactionId, boolean readOnly, Map<String, String> attributes)
       throws TransactionException {
-    Snapshot snapshot = new Snapshot(transactionId, tableMetadataManager, parallelExecutor);
+    Snapshot snapshot =
+        new Snapshot(transactionId, tableMetadataManager, parallelExecutor, collationComparator);
     // The isolation level can be overridden per transaction via the transaction-isolation
     // attribute; otherwise it falls back to the participant's configured default.
     Isolation isolation =