Skip to content

Commit fad32da

Browse files
brfrn169brfrn169
committed
Prevent overwriting read set in Consensus Commit
1 parent aadc7f8 commit fad32da

File tree

3 files changed

+131
-9
lines changed

3 files changed

+131
-9
lines changed

‎core/src/main/java/com/scalar/db/transaction/consensuscommit/CrudHandler.java‎

Lines changed: 5 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -164,11 +164,10 @@ Optional<TransactionResult> read(@Nullable Snapshot.Key key, Get get) throws Cru
164164
}
165165

166166
if (result.isPresent() || get.getConjunctions().isEmpty()) {
167-
// Keep the read set latest to create before image by using the latest record (result)
168-
// because another conflicting transaction might have updated the record after this
169-
// transaction read it first. However, we update it only if a get operation has no
170-
// conjunction or the result exists. This is because we don’t know whether the record
171-
// actually exists or not due to the conjunction.
167+
// We put the result into the read set only if a get operation has no conjunction or the
168+
// result exists. This is because we don’t know whether the record actually exists or not
169+
// due to the conjunction.
170+
172171
if (key != null) {
173172
putIntoReadSetInSnapshot(key, result);
174173
} else {
@@ -277,9 +276,6 @@ private Optional<TransactionResult> processScanResult(
277276
}
278277

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

@@ -319,7 +315,7 @@ public void closeScanners() throws CrudException {
319315

320316
private void putIntoReadSetInSnapshot(Snapshot.Key key, Optional<TransactionResult> result) {
321317
// In read-only mode, we don't need to put the result into the read set
322-
if (!readOnly) {
318+
if (!readOnly && !snapshot.containsKeyInReadSet(key)) {
323319
snapshot.putIntoReadSet(key, result);
324320
}
325321
}

‎core/src/test/java/com/scalar/db/transaction/consensuscommit/CrudHandlerTest.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -541,6 +541,45 @@ public void get_ForNonExistingTable_ShouldThrowIllegalArgumentException()
541541
assertThatThrownBy(() -> handler.get(get)).isInstanceOf(IllegalArgumentException.class);
542542
}
543543

544+
@Test
545+
public void get_DifferentGetButSameRecordReturned_ShouldNotOverwriteReadSet()
546+
throws ExecutionException, CrudException {
547+
// Arrange
548+
Get get1 = prepareGet();
549+
Get get2 = Get.newBuilder(get1).where(column(ANY_NAME_3).isEqualToText(ANY_TEXT_3)).build();
550+
Get getForStorage1 = toGetForStorageFrom(get1);
551+
Get getForStorage2 =
552+
Get.newBuilder(get2)
553+
.clearProjections()
554+
.projections(TRANSACTION_TABLE_METADATA.getAfterImageColumnNames())
555+
.clearConditions()
556+
.where(column(ANY_NAME_3).isEqualToText(ANY_TEXT_3))
557+
.or(column(Attribute.BEFORE_PREFIX + ANY_NAME_3).isEqualToText(ANY_TEXT_3))
558+
.consistency(Consistency.LINEARIZABLE)
559+
.build();
560+
Result result = prepareResult(TransactionState.COMMITTED);
561+
Optional<TransactionResult> expected = Optional.of(new TransactionResult(result));
562+
Snapshot.Key key = new Snapshot.Key(getForStorage1);
563+
when(snapshot.getResult(any(), any())).thenReturn(expected).thenReturn(expected);
564+
when(snapshot.containsKeyInReadSet(key)).thenReturn(false).thenReturn(true);
565+
when(storage.get(any())).thenReturn(Optional.of(result));
566+
567+
// Act
568+
Optional<Result> results1 = handler.get(get1);
569+
Optional<Result> results2 = handler.get(get2);
570+
571+
// Assert
572+
assertThat(results1)
573+
.isEqualTo(
574+
Optional.of(
575+
new FilteredResult(
576+
expected.get(), Collections.emptyList(), TABLE_METADATA, false)));
577+
assertThat(results2).isEqualTo(results1);
578+
verify(storage).get(getForStorage1);
579+
verify(storage).get(getForStorage2);
580+
verify(snapshot).putIntoReadSet(key, expected);
581+
}
582+
544583
@ParameterizedTest
545584
@EnumSource(ScanType.class)
546585
void scanOrGetScanner_ResultGivenFromStorage_ShouldUpdateSnapshotAndReturn(ScanType scanType)

‎integration-test/src/main/java/com/scalar/db/transaction/consensuscommit/ConsensusCommitSpecificIntegrationTestBase.java‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package com.scalar.db.transaction.consensuscommit;
22

33
import static com.scalar.db.api.ConditionBuilder.column;
4+
import static com.scalar.db.api.ConditionBuilder.updateIf;
45
import static org.assertj.core.api.Assertions.assertThat;
56
import static org.assertj.core.api.Assertions.assertThatCode;
67
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -45,6 +46,7 @@
4546
import com.scalar.db.exception.transaction.CrudConflictException;
4647
import com.scalar.db.exception.transaction.CrudException;
4748
import com.scalar.db.exception.transaction.PreparationConflictException;
49+
import com.scalar.db.exception.transaction.RollbackException;
4850
import com.scalar.db.exception.transaction.TransactionException;
4951
import com.scalar.db.exception.transaction.UnknownTransactionStatusException;
5052
import com.scalar.db.io.DataType;
@@ -5893,6 +5895,91 @@ public void getScanner_InReadOnlyMode_WithSerializable_ShouldNotThrowAnyExceptio
58935895
assertThat(results.get(1).getInt(BALANCE)).isEqualTo(INITIAL_BALANCE);
58945896
}
58955897

5898+
@Test
5899+
public void
5900+
commit_ConflictingExternalUpdate_DifferentGetButSameRecordReturned_ShouldThrowCommitConflictExceptionAndPreserveExternalChanges()
5901+
throws UnknownTransactionStatusException, CrudException, RollbackException {
5902+
// Arrange
5903+
manager.insert(
5904+
Insert.newBuilder()
5905+
.namespace(namespace1)
5906+
.table(TABLE_1)
5907+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5908+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5909+
.intValue(BALANCE, INITIAL_BALANCE)
5910+
.build());
5911+
5912+
// Act Assert
5913+
DistributedTransaction transaction = manager.begin();
5914+
5915+
// Retrieve the record
5916+
Optional<Result> result =
5917+
transaction.get(
5918+
Get.newBuilder()
5919+
.namespace(namespace1)
5920+
.table(TABLE_1)
5921+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5922+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5923+
.build());
5924+
5925+
assertThat(result).isPresent();
5926+
assertThat(result.get().getInt(ACCOUNT_ID)).isEqualTo(0);
5927+
assertThat(result.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
5928+
assertThat(result.get().getInt(BALANCE)).isEqualTo(INITIAL_BALANCE);
5929+
5930+
// Update the balance of the record
5931+
transaction.update(
5932+
Update.newBuilder()
5933+
.namespace(namespace1)
5934+
.table(TABLE_1)
5935+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5936+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5937+
.condition(updateIf(column(BALANCE).isEqualToInt(INITIAL_BALANCE)).build())
5938+
.intValue(BALANCE, 100)
5939+
.build());
5940+
5941+
// Update the balance of the record by another transaction
5942+
manager.update(
5943+
Update.newBuilder()
5944+
.namespace(namespace1)
5945+
.table(TABLE_1)
5946+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5947+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5948+
.intValue(BALANCE, 200)
5949+
.build());
5950+
5951+
// Retrieve the record again, but use a different Get object (with a where clause)
5952+
result =
5953+
transaction.get(
5954+
Get.newBuilder()
5955+
.namespace(namespace1)
5956+
.table(TABLE_1)
5957+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5958+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5959+
.where(column(BALANCE).isEqualToInt(200))
5960+
.build());
5961+
5962+
assertThat(result).isNotPresent();
5963+
5964+
assertThatThrownBy(transaction::commit).isInstanceOf(CommitConflictException.class);
5965+
transaction.rollback();
5966+
5967+
// Assert
5968+
result =
5969+
manager.get(
5970+
Get.newBuilder()
5971+
.namespace(namespace1)
5972+
.table(TABLE_1)
5973+
.partitionKey(Key.ofInt(ACCOUNT_ID, 0))
5974+
.clusteringKey(Key.ofInt(ACCOUNT_TYPE, 0))
5975+
.build());
5976+
5977+
assertThat(result).isPresent();
5978+
assertThat(result.get().getInt(ACCOUNT_ID)).isEqualTo(0);
5979+
assertThat(result.get().getInt(ACCOUNT_TYPE)).isEqualTo(0);
5980+
assertThat(result.get().getInt(BALANCE)).isEqualTo(200);
5981+
}
5982+
58965983
@Test
58975984
public void manager_get_GetGivenForCommittedRecord_WithSerializable_ShouldReturnRecord()
58985985
throws TransactionException {

0 commit comments

Comments
 (0)