Skip to content

Prevent overwriting read set in Consensus Commit #2797

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

Merged
merged 1 commit into from
Jun 19, 2025
Merged

Prevent overwriting read set in Consensus Commit #2797

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
14 changes: 5 additions & 9 deletions core/src/main/java/com/scalar/db/transaction/consensuscommit/CrudHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,11 +164,10 @@ Optional<TransactionResult> read(@Nullable Snapshot.Key key, Get get) throws Cru
}

if (result.isPresent() || get.getConjunctions().isEmpty()) {
// Keep the read set latest to create before image by using the latest record (result)
// because another conflicting transaction might have updated the record after this
// transaction read it first. However, we update it only if a get operation has no
Comment on lines -167 to -169
Copy link
Contributor

@jnmt jnmt Jun 19, 2025

Choose a reason for hiding this comment

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

Sorry, my bad. This part was a completely wrong idea. Probably, I was trying to avoid aborts as much as possible, but paid little attention to the case for reading a record multiple times. I need to think about how we can avoid these kinds of concurrency bugs…

// conjunction or the result exists. This is because we don’t know whether the record
// actually exists or not due to the conjunction.
// We put the result into the read set only if a get operation has no conjunction or the
// result exists. This is because we don’t know whether the record actually exists or not
// due to the conjunction.

if (key != null) {
putIntoReadSetInSnapshot(key, result);
} else {
Expand Down Expand Up @@ -277,9 +276,6 @@ private Optional<TransactionResult> processScanResult(
}

if (ret.isPresent()) {
// We always update the read set to create before image by using the latest record (result)
// because another conflicting transaction might have updated the record after this
// transaction read it first.
putIntoReadSetInSnapshot(key, ret);
}

Expand Down Expand Up @@ -319,7 +315,7 @@ public void closeScanners() throws CrudException {

private void putIntoReadSetInSnapshot(Snapshot.Key key, Optional<TransactionResult> result) {
// In read-only mode, we don't need to put the result into the read set
if (!readOnly) {
if (!readOnly && !snapshot.containsKeyInReadSet(key)) {
Copy link
Contributor

@komamitsu komamitsu Jun 19, 2025

Choose a reason for hiding this comment

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

IIUC, we can use the latest result if read-modify-write hasn't occurred on the key, like as follows?

    if (!readOnly && !(snapshot.containsKeyInReadSet(key) && snapshot.containsKeyInWriteOrDeleteSet(key)) {

This question is mainly for confirmation and the current code looks good to me, though.

Copy link
Collaborator Author

@brfrn169 brfrn169 Jun 19, 2025

Choose a reason for hiding this comment

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

Sorry, what do you mean by read-modify-write? You mean CAS operations in the prepare-records phase, right?

Copy link
Contributor

@komamitsu komamitsu Jun 19, 2025

Choose a reason for hiding this comment

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

Sorry for the unclear comment. My question was as follows:

  • With this PR, the read-set isn't updated if the key is read
  • I think we can update the read-set even if the key is read, as long as the key isn't written or deleted in the transaction
  • For instance
    • Record A has the key = 100 and value = "one"
    • Transaction 1 reads the key and the value is "one", and puts it into the read-set
    • Transaction 2 updates the value of the record to "two" and commits it
    • Transaction 1 reads the key with different GET and the value is "two", and puts it into the read-set since there is no write operation on the key in Transaction 1
    • Transaction 1 updates the value of the record to "three" based on the read value "two"
    • In this case, no anomaly doesn't occur even if we use the latest result as a read-set
      • If Transaction 1 updates the record before the second GET, the second GET's result shouldn't be put into the read-set to avoid lost update anomaly

Copy link
Contributor

@jnmt jnmt Jun 19, 2025

Choose a reason for hiding this comment

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

Good point. We might be able to rescue such a transaction, but I'm really not sure if it's good for our snapshot definition. One approach is to make it a safer side (like the current PR) for now, and then improve it after the deep discussion and confirmation.

Copy link
Collaborator Author

@brfrn169 brfrn169 Jun 19, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

@komamitsu Indeed, we might be able to rescue such a transaction. However, since we don’t have enough time to test it, let’s make it the safer side for now as @jnmt suggested. Thanks!

snapshot.putIntoReadSet(key, result);
}
}
Expand Down
39 changes: 39 additions & 0 deletions core/src/test/java/com/scalar/db/transaction/consensuscommit/CrudHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -541,6 +541,45 @@ public void get_ForNonExistingTable_ShouldThrowIllegalArgumentException()
assertThatThrownBy(() -> handler.get(get)).isInstanceOf(IllegalArgumentException.class);
}

@Test
public void get_DifferentGetButSameRecordReturned_ShouldNotOverwriteReadSet()
throws ExecutionException, CrudException {
// Arrange
Get get1 = prepareGet();
Get get2 = Get.newBuilder(get1).where(column(ANY_NAME_3).isEqualToText(ANY_TEXT_3)).build();
Get getForStorage1 = toGetForStorageFrom(get1);
Get getForStorage2 =
Get.newBuilder(get2)
.clearProjections()
.projections(TRANSACTION_TABLE_METADATA.getAfterImageColumnNames())
.clearConditions()
.where(column(ANY_NAME_3).isEqualToText(ANY_TEXT_3))
.or(column(Attribute.BEFORE_PREFIX + ANY_NAME_3).isEqualToText(ANY_TEXT_3))
.consistency(Consistency.LINEARIZABLE)
.build();
Result result = prepareResult(TransactionState.COMMITTED);
Optional<TransactionResult> expected = Optional.of(new TransactionResult(result));
Snapshot.Key key = new Snapshot.Key(getForStorage1);
when(snapshot.getResult(any(), any())).thenReturn(expected).thenReturn(expected);
when(snapshot.containsKeyInReadSet(key)).thenReturn(false).thenReturn(true);
when(storage.get(any())).thenReturn(Optional.of(result));

// Act
Optional<Result> results1 = handler.get(get1);
Optional<Result> results2 = handler.get(get2);

// Assert
assertThat(results1)
.isEqualTo(
Optional.of(
new FilteredResult(
expected.get(), Collections.emptyList(), TABLE_METADATA, false)));
assertThat(results2).isEqualTo(results1);
verify(storage).get(getForStorage1);
verify(storage).get(getForStorage2);
verify(snapshot).putIntoReadSet(key, expected);
}

@ParameterizedTest
@EnumSource(ScanType.class)
void scanOrGetScanner_ResultGivenFromStorage_ShouldUpdateSnapshotAndReturn(ScanType scanType)
Expand Down
87 changes: 87 additions & 0 deletions ...com/scalar/db/transaction/consensuscommit/ConsensusCommitSpecificIntegrationTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package com.scalar.db.transaction.consensuscommit;

import static com.scalar.db.api.ConditionBuilder.column;
import static com.scalar.db.api.ConditionBuilder.updateIf;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
Expand Down Expand Up @@ -45,6 +46,7 @@
import com.scalar.db.exception.transaction.CrudConflictException;
import com.scalar.db.exception.transaction.CrudException;
import com.scalar.db.exception.transaction.PreparationConflictException;
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.DataType;
Expand Down Expand Up @@ -5893,6 +5895,91 @@ public void getScanner_InReadOnlyMode_WithSerializable_ShouldNotThrowAnyExceptio
assertThat(results.get(1).getInt(BALANCE)).isEqualTo(INITIAL_BALANCE);
}

@Test
public void
commit_ConflictingExternalUpdate_DifferentGetButSameRecordReturned_ShouldThrowCommitConflictExceptionAndPreserveExternalChanges()
throws UnknownTransactionStatusException, CrudException, RollbackException {
// Arrange
manager.insert(
Insert.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.intValue(BALANCE, INITIAL_BALANCE)
.build());

// Act Assert
DistributedTransaction transaction = manager.begin();

// Retrieve the record
Optional<Result> result =
transaction.get(
Get.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.build());

assertThat(result).isPresent();
assertThat(result.get().getInt(ACCOUNT_ID)).isEqualTo(0);
assertThat(result.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
assertThat(result.get().getInt(BALANCE)).isEqualTo(INITIAL_BALANCE);

// Update the balance of the record
transaction.update(
Update.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.condition(updateIf(column(BALANCE).isEqualToInt(INITIAL_BALANCE)).build())
.intValue(BALANCE, 100)
.build());

// Update the balance of the record by another transaction
manager.update(
Update.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.intValue(BALANCE, 200)
.build());

// Retrieve the record again, but use a different Get object (with a where clause)
Copy link
Contributor

@komamitsu komamitsu Jun 19, 2025

Choose a reason for hiding this comment

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

👍

result =
transaction.get(
Get.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.where(column(BALANCE).isEqualToInt(200))
.build());

assertThat(result).isNotPresent();

assertThatThrownBy(transaction::commit).isInstanceOf(CommitConflictException.class);
transaction.rollback();

// Assert
result =
manager.get(
Get.newBuilder()
.namespace(namespace1)
.table(TABLE_1)
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
.build());

assertThat(result).isPresent();
assertThat(result.get().getInt(ACCOUNT_ID)).isEqualTo(0);
assertThat(result.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
assertThat(result.get().getInt(BALANCE)).isEqualTo(200);
}

@Test
public void manager_get_GetGivenForCommittedRecord_WithSerializable_ShouldReturnRecord()
throws TransactionException {
Expand Down
Loading