Add further optimizations for Consensus Commit #2764
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -51,6 +51,11 @@ public class CrudHandler { | |
| // Whether the transaction is in read-only mode or not. | ||
| private final boolean readOnly; | ||
|
|
||
| // Whether the transaction is in single-operation mode or not. Single-operation mode refers to | ||
| // executing a CRUD operation directly through `DistributedTransactionManager` without explicitly | ||
| // beginning a transaction. | ||
| private final boolean singleOperation; | ||
|
||
|
|
||
| private final List<ConsensusCommitScanner> scanners = new ArrayList<>(); | ||
|
|
||
| @SuppressFBWarnings("EI_EXPOSE_REP2") | ||
|
|
@@ -60,14 +65,16 @@ public CrudHandler( | |
| TransactionTableMetadataManager tableMetadataManager, | ||
| boolean isIncludeMetadataEnabled, | ||
| ParallelExecutor parallelExecutor, | ||
| boolean readOnly, | ||
| boolean singleOperation) { | ||
| this.storage = checkNotNull(storage); | ||
| this.snapshot = checkNotNull(snapshot); | ||
| this.tableMetadataManager = tableMetadataManager; | ||
| this.isIncludeMetadataEnabled = isIncludeMetadataEnabled; | ||
| this.mutationConditionsValidator = new MutationConditionsValidator(snapshot.getId()); | ||
| this.parallelExecutor = parallelExecutor; | ||
| this.readOnly = readOnly; | ||
| this.singleOperation = singleOperation; | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
|
|
@@ -78,14 +85,16 @@ public CrudHandler( | |
| boolean isIncludeMetadataEnabled, | ||
| MutationConditionsValidator mutationConditionsValidator, | ||
| ParallelExecutor parallelExecutor, | ||
| boolean readOnly, | ||
| boolean singleOperation) { | ||
| this.storage = checkNotNull(storage); | ||
| this.snapshot = checkNotNull(snapshot); | ||
| this.tableMetadataManager = tableMetadataManager; | ||
| this.isIncludeMetadataEnabled = isIncludeMetadataEnabled; | ||
| this.mutationConditionsValidator = mutationConditionsValidator; | ||
| this.parallelExecutor = parallelExecutor; | ||
| this.readOnly = readOnly; | ||
| this.singleOperation = singleOperation; | ||
| } | ||
|
|
||
| public Optional<Result> get(Get originalGet) throws CrudException { | ||
|
|
@@ -102,11 +111,17 @@ public Optional<Result> get(Get originalGet) throws CrudException { | |
| key = new Snapshot.Key(get); | ||
| } | ||
|
|
||
| if (isSnapshotReadRequired()) { | ||
| readUnread(key, get); | ||
| return snapshot | ||
| .getResult(key, get) | ||
| .map(r -> new FilteredResult(r, originalProjections, metadata, isIncludeMetadataEnabled)); | ||
| } else { | ||
| Optional<TransactionResult> result = read(key, get); | ||
| return snapshot | ||
| .mergeResult(key, result, get.getConjunctions()) | ||
| .map(r -> new FilteredResult(r, originalProjections, metadata, isIncludeMetadataEnabled)); | ||
| } | ||
| } | ||
|
|
||
| // Only for a Get with index, the argument `key` is null | ||
|
|
@@ -120,7 +135,7 @@ void readUnread(@Nullable Snapshot.Key key, Get get) throws CrudException { | |
| // Although this class is not thread-safe, this method is actually thread-safe, so we call it | ||
| // concurrently in the implicit pre-read | ||
| @VisibleForTesting | ||
| Optional<TransactionResult> read(@Nullable Snapshot.Key key, Get get) throws CrudException { | ||
| Optional<TransactionResult> result = getFromStorage(get); | ||
| if (!result.isPresent() || result.get().isCommitted()) { | ||
| if (result.isPresent() || get.getConjunctions().isEmpty()) { | ||
|
|
@@ -142,8 +157,8 @@ void read(@Nullable Snapshot.Key key, Get get) throws CrudException { | |
| } | ||
| } | ||
| } | ||
| putIntoGetSetInSnapshot(get, result); | ||
| return result; | ||
| } | ||
| throw new UncommittedRecordException( | ||
| get, | ||
|
|
@@ -204,7 +219,7 @@ private LinkedHashMap<Snapshot.Key, TransactionResult> scanInternal(Scan scan) | |
| } | ||
| } | ||
|
|
||
| putIntoScanSetInSnapshot(scan, results); | ||
|
|
||
| return results; | ||
| } | ||
|
|
@@ -263,9 +278,43 @@ private void putIntoReadSetInSnapshot(Snapshot.Key key, Optional<TransactionResu | |
| } | ||
| } | ||
|
|
||
| private boolean isSnapshotReadRequired() { | ||
| // In single-operation mode, we don't need snapshot read | ||
| return !singleOperation; | ||
| } | ||
|
|
||
| private boolean isValidationOrSnapshotReadRequired() { | ||
| return snapshot.isValidationRequired() || isSnapshotReadRequired(); | ||
| } | ||
|
|
||
| private void putIntoGetSetInSnapshot(Get get, Optional<TransactionResult> result) { | ||
| // If neither validation nor snapshot read is required, we don't need to put the result into | ||
| // the get set | ||
| if (isValidationOrSnapshotReadRequired()) { | ||
| snapshot.putIntoGetSet(get, result); | ||
| } | ||
| } | ||
|
|
||
| private void putIntoScanSetInSnapshot( | ||
| Scan scan, LinkedHashMap<Snapshot.Key, TransactionResult> results) { | ||
| // If neither validation nor snapshot read is required, we don't need to put the results into | ||
| // the scan set | ||
| if (isValidationOrSnapshotReadRequired()) { | ||
| snapshot.putIntoScanSet(scan, results); | ||
| } | ||
| } | ||
|
|
||
| private void putIntoScannerSetInSnapshot( | ||
| Scan scan, LinkedHashMap<Snapshot.Key, TransactionResult> results) { | ||
| // if validation is not required, we don't need to put the results into the scanner set | ||
| if (snapshot.isValidationRequired()) { | ||
| snapshot.putIntoScannerSet(scan, results); | ||
| } | ||
| } | ||
|
|
||
| private void verifyNoOverlap(Scan scan, Map<Snapshot.Key, TransactionResult> results) { | ||
| // In either read-only mode or single-operation mode, we don't need to verify the overlap | ||
| if (!readOnly && !singleOperation) { | ||
| snapshot.verifyNoOverlap(scan, results); | ||
| } | ||
| } | ||
|
Collaborator
Author
There was a problem hiding this comment. Here is the main part of the optimizations. The optimizations can be summarized as follows:
Contributor
There was a problem hiding this comment. If a single-crud operation is a self-join reading the same record more than twice, is it OK not to use snapshot read?
Collaborator
Author
There was a problem hiding this comment. The single-operation mode here means the mode for transactions that are executed without explicitly beginning a transaction, such as the following methods: DistributedTransactionManger transactionManager = ...;
Optional<Result> result = transactionManager.get(Get.newBuilder()...);
List<Result> results = transactionManager.scan(Scan.newBuilder()...);
Scanner scanner = transactionManager.getScanner(Scan.newBuilder()...);
transactionManager.insert(Insert.newBuilder()...);
transactionManager.upsert(Upsert.newBuilder()...);
transactionManager.update(Update.newBuilder()...);
transactionManager.delete(Delete.newBuilder()...);
transactionManager.mutate(Arrays.asList(Delete.newBuilder()..., Update.newBuilder()...));In these cases, the transaction is committed immediately after the operation is done. So we don't need snapshot reads in such cases. Does this answer your question?
Contributor
There was a problem hiding this comment. @brfrn169 Probably, I am confused with single-crud operation transaction and single-operation mode.
Collaborator
Author
There was a problem hiding this comment. Ah, okay. This isn’t related to single CRUD operation transactions — it’s specific to Consensus Commit. Do you have any suggestions for what we should call transactions that are executed without explicitly beginning a transaction? |
||
|
|
@@ -432,7 +481,7 @@ private class ConsensusCommitStorageScanner extends AbstractTransactionCrudOpera | |
| private final List<String> originalProjections; | ||
| private final Scanner scanner; | ||
|
|
||
| @Nullable private final LinkedHashMap<Snapshot.Key, TransactionResult> results; | ||
| private final AtomicBoolean fullyScanned = new AtomicBoolean(); | ||
| private final AtomicBoolean closed = new AtomicBoolean(); | ||
|
|
||
|
|
@@ -441,6 +490,14 @@ public ConsensusCommitStorageScanner(Scan scan, List<String> originalProjections | |
| this.scan = scan; | ||
| this.originalProjections = originalProjections; | ||
| scanner = scanFromStorage(scan); | ||
|
|
||
| if (isValidationOrSnapshotReadRequired()) { | ||
| results = new LinkedHashMap<>(); | ||
| } else { | ||
| // If neither validation nor snapshot read is required, we don't need to put the results | ||
| // into the scan set | ||
| results = null; | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -456,7 +513,10 @@ public Optional<Result> one() throws CrudException { | |
| Snapshot.Key key = new Snapshot.Key(scan, r.get()); | ||
| TransactionResult result = new TransactionResult(r.get()); | ||
| processScanResult(key, scan, result); | ||
|
|
||
| if (results != null) { | ||
| results.put(key, result); | ||
| } | ||
|
|
||
| TableMetadata metadata = getTableMetadata(scan); | ||
| return Optional.of( | ||
|
|
@@ -499,10 +559,10 @@ public void close() { | |
| if (fullyScanned.get()) { | ||
| // If the scanner is fully scanned, we can treat it as a normal scan, and put the results | ||
| // into the scan set | ||
| putIntoScanSetInSnapshot(scan, results); | ||
| } else { | ||
| // If the scanner is not fully scanned, put the results into the scanner set | ||
| putIntoScannerSetInSnapshot(scan, results); | ||
| } | ||
|
|
||
| verifyNoOverlap(scan, results); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Fixed in 042d1e1. Thanks!