Skip to content

Collation 3/4: follow the collation in Consensus Commit layer #3844

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Torch3333 wants to merge 1 commit into from
+2,685 −483
Open

Collation 3/4: follow the collation in Consensus Commit layer #3844

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions core/src/main/java/com/scalar/db/common/checker/ConditionChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,12 @@ public void visit(DeleteIfExists condition) {

private void checkExpressions(List<ConditionalExpression> expressions) {
for (ConditionalExpression expression : expressions) {
if (expression.getOperator() == Operator.LIKE
|| expression.getOperator() == Operator.NOT_LIKE) {
// No storage evaluates pattern matching in a conditional write.
isValid = false;
break;
}
Comment on lines +72 to +77

@brfrn169 brfrn169 Sep 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The LIKE/NOT_LIKE rejection added here doesn't take effect in Consensus Commit. ConsensusCommitOperationChecker calls ConditionChecker.check() for Put and Delete but discards the result:

Since ConditionChecker has no side effects, these calls currently do nothing. A LIKE condition that bypasses the builder check (e.g., via ConditionBuilder.putIf(List) or deleteIf(List)) passes the checker and, when the target record exists, hits default: throw new AssertionError() in MutationConditionsValidator.shouldMutate(). Other conditions that the storage-side OperationChecker rejects, such as a nonexistent column or a type mismatch, aren't rejected up front in Consensus Commit either.

This bug isn't introduced by this PR. The result has been discarded since the call was added in #899, and the same code is on 3 and 3.16 through 3.19. Could you fix it in a separate PR against master? The fix should be backported to all the supported release branches, which is easier as a standalone change than as part of this collation stack.

Some notes for that PR:

  • Throwing IllegalArgumentException when check() returns false, as OperationChecker.checkCondition() does with CoreError.OPERATION_CHECK_ERROR_CONDITION, would align Consensus Commit with the storage API and JDBC transactions.
  • This is a behavior change. Conditions on primary-key columns and comparisons against a null value (e.g., isEqualToText(null)) are currently accepted and evaluated in memory; they would start to be rejected.
  • ConsensusCommitOperationCheckerTest uses an unstubbed ConditionChecker mock, which returns false, so the existing tests that assert no exception would need check() stubbed to return true.

if (expression.getOperator() == Operator.IS_NULL
|| expression.getOperator() == Operator.IS_NOT_NULL) {
// the value must be null if the operator is 'is null' or `is not null`
Expand Down
23 changes: 17 additions & 6 deletions core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand All @@ -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();
Expand All @@ -113,7 +117,8 @@ public ConsensusCommitManager(
tableMetadataManager,
virtualTableInfoManager,
storageInfoProvider,
config.isIncludeMetadataEnabled());
config.isIncludeMetadataEnabled(),
databaseConfig.getCollation());

ConsensusCommitUtils.warnIfBeforeIndexesAreMissing(admin, config);
}
Expand All @@ -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();
Expand All @@ -153,7 +160,8 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
tableMetadataManager,
virtualTableInfoManager,
storageInfoProvider,
config.isIncludeMetadataEnabled());
config.isIncludeMetadataEnabled(),
databaseConfig.getCollation());

ConsensusCommitUtils.warnIfBeforeIndexesAreMissing(admin, config);
}
Expand Down Expand Up @@ -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());
Expand All @@ -195,7 +204,8 @@ protected ConsensusCommitManager(DatabaseConfig databaseConfig) {
tableMetadataManager,
virtualTableInfoManager,
storageInfoProvider,
config.isIncludeMetadataEnabled());
config.isIncludeMetadataEnabled(),
databaseConfig.getCollation());
}

// `groupCommitter` must be set before calling this method.
Expand Down Expand Up @@ -309,7 +319,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);
Expand Down
48 changes: 47 additions & 1 deletion .../main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitOperationChecker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand All @@ -36,16 +38,35 @@ 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;
}

private void throwIfLikeConditionUnderIcuCollation(Selection selection) {

@brfrn169 brfrn169 Sep 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: The other private helpers in this class, including throwIfKeyedMutationAtomicityUnitUnderIcuCollation added in this PR, are placed after the public check methods. Could you move this one there as well?

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()));
Comment on lines +63 to +66

@brfrn169 brfrn169 Sep 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Question: Could you explain why LIKE/NOT_LIKE can't be supported under ICU?

}
}
}
}

/**
Expand All @@ -58,6 +79,7 @@ public ConsensusCommitOperationChecker(
*/
public void check(Get get, TransactionContext context) throws ExecutionException {
throwIfOperationForVirtualTableButNotConsistentVirtualTableReadStorage(get);
throwIfLikeConditionUnderIcuCollation(get);

TransactionTableMetadata metadata =
getTransactionTableMetadata(transactionTableMetadataManager, get);
Expand Down Expand Up @@ -115,6 +137,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);
Expand Down Expand Up @@ -201,6 +224,7 @@ public void check(Scan scan, TransactionContext context) throws ExecutionExcepti
*/
public void check(Mutation mutation) throws ExecutionException {
throwIfOperationForVirtualTableButNotConsistentVirtualTableReadStorage(mutation);
throwIfKeyedMutationAtomicityUnitUnderIcuCollation(mutation);

if (mutation instanceof Put) {
check((Put) mutation);
Expand Down Expand Up @@ -274,6 +298,28 @@ ConditionChecker createConditionChecker(TableMetadata tableMetadata) {
return new ConditionChecker(tableMetadata);
}

private void throwIfKeyedMutationAtomicityUnitUnderIcuCollation(Mutation mutation)
throws ExecutionException {
if (collation != Collation.ICU) {
return;
}
assert mutation.forNamespace().isPresent();
StorageInfo storageInfo = storageInfoProvider.getStorageInfo(mutation.forNamespace().get());
StorageInfo.MutationAtomicityUnit unit = storageInfo.getMutationAtomicityUnit();
if (unit == StorageInfo.MutationAtomicityUnit.RECORD
|| unit == StorageInfo.MutationAtomicityUnit.PARTITION) {
// No storage with a record or partition atomicity supports ICU collation
// If we support such storage in the future, we need to make the MutationsGrouper collation
// aware
Comment on lines +311 to +313
throw new UnsupportedOperationException(
Comment on lines +311 to +314

@brfrn169 brfrn169 Sep 15, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This branch is effectively dead code: every storage whose mutation atomicity unit is RECORD or PARTITION rejects the ICU collation when it is created, and multi-storage doesn't allow overriding the collation per storage. How about throwing AssertionError instead and writing down why it's unreachable?

Suggested change
// No storage with a record or partition atomicity supports ICU collation
// If we support such storage in the future, we need to make the MutationsGrouper collation
// aware
throw new UnsupportedOperationException(
// Unreachable: every storage whose mutation atomicity unit is RECORD or PARTITION rejects
// the ICU collation when it is created (CoreError.COLLATION_ICU_NOT_SUPPORTED_BY_STORAGE),
// and multi-storage doesn't allow overriding the collation per storage. If such a storage
// supports the ICU collation in the future, revisit MutationsGrouper, which compares keys
// byte-exactly.
throw new AssertionError(

checkForMutation_WithKeyedMutationAtomicityUnitUnderIcuCollation_ShouldThrowUnsupportedOperationException would need to be updated as well.

"The ICU collation is not supported for a storage that applies mutations atomically only "
+ "within a record or a partition. Storage: "
+ storageInfo.getStorageName()
+ "; Mutation atomicity unit: "
+ unit);
}
}

private void throwIfOperationForVirtualTableButNotConsistentVirtualTableReadStorage(
Operation operation) throws ExecutionException {
assert operation.forNamespace().isPresent();
Expand Down
16 changes: 12 additions & 4 deletions core/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitParticipant.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<>();

Expand All @@ -101,14 +103,16 @@ 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(
Expand All @@ -124,7 +128,8 @@ public ConsensusCommitParticipant(DatabaseConfig databaseConfig) {
tableMetadataManager,
virtualTableInfoManager,
storageInfoProvider,
config.isIncludeMetadataEnabled());
config.isIncludeMetadataEnabled(),
databaseConfig.getCollation());
}

@VisibleForTesting
Expand All @@ -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;
Expand All @@ -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) {
Expand All @@ -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 =
Expand Down
Loading