You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "rpuch (via GitHub)" <gi...@apache.org> on 2023/01/23 10:52:42 UTC
[GitHub] [ignite-3] rpuch commented on a diff in pull request #1556: IGNITE-18593 Get rid of MvPartitionStorage and TxStateStorage in PartitionAccess
rpuch commented on code in PR #1556:
URL: https://github.com/apache/ignite-3/pull/1556#discussion_r1083873554
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionAccess.java:
##########
@@ -32,27 +41,111 @@ public interface PartitionAccess {
PartitionKey partitionKey();
/**
- * Returns the multi-versioned partition storage.
+ * Destroys and recreates the multi-versioned partition storage.
+ *
+ * @return Future that will complete when the partition is recreated.
+ * @throws StorageException If an error has occurred during the partition destruction.
*/
- MvPartitionStorage mvPartitionStorage();
+ CompletableFuture<Void> reCreateMvPartitionStorage() throws StorageException;
/**
- * Returns transaction state storage for the partition.
+ * Destroys and recreates the multi-versioned partition storage.
Review Comment:
```suggestion
* Destroys and recreates the TX state partition storage.
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionAccess.java:
##########
@@ -32,27 +41,111 @@ public interface PartitionAccess {
PartitionKey partitionKey();
/**
- * Returns the multi-versioned partition storage.
+ * Destroys and recreates the multi-versioned partition storage.
+ *
+ * @return Future that will complete when the partition is recreated.
+ * @throws StorageException If an error has occurred during the partition destruction.
*/
- MvPartitionStorage mvPartitionStorage();
+ CompletableFuture<Void> reCreateMvPartitionStorage() throws StorageException;
/**
- * Returns transaction state storage for the partition.
+ * Destroys and recreates the multi-versioned partition storage.
+ *
+ * @throws StorageException If an error has occurred during transaction state storage for the partition destruction.
Review Comment:
```suggestion
* @throws StorageException If an error has occurred during destruction of the transaction state storage for the partition.
```
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/outgoing/OutgoingSnapshotCommonTest.java:
##########
@@ -76,12 +64,13 @@ void returnsKeyFromStorage() {
@Test
void sendsSnapshotMeta() {
- when(mvPartitionStorage.lastAppliedIndex()).thenReturn(100L);
- lenient().when(mvPartitionStorage.lastAppliedTerm()).thenReturn(3L);
- when(txStateStorage.lastAppliedIndex()).thenReturn(100L);
- lenient().when(txStateStorage.lastAppliedTerm()).thenReturn(3L);
+ lenient().when(partitionAccess.minLastAppliedIndex()).thenReturn(90L);
Review Comment:
These `lenient()`s seem redundant and should be dropped
##########
modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorageFactoryTest.java:
##########
@@ -44,24 +39,18 @@
*/
@ExtendWith(MockitoExtension.class)
public class PartitionSnapshotStorageFactoryTest {
- private final MvPartitionStorage mvPartitionStorage = new TestMvPartitionStorage(0);
- private final TxStateStorage txStateStorage = new TestTxStateStorage();
-
@Mock
private PartitionAccess partitionAccess;
- @BeforeEach
- void configureMocks() {
- when(partitionAccess.mvPartitionStorage()).thenReturn(mvPartitionStorage);
- when(partitionAccess.txStatePartitionStorage()).thenReturn(txStateStorage);
- }
-
@Test
void testForChoosingMinimumAppliedIndexForMeta() {
- mvPartitionStorage.lastApplied(10L, 2L);
- txStateStorage.lastApplied(5L, 1L);
+ lenient().when(partitionAccess.minLastAppliedIndex()).thenReturn(5L);
Review Comment:
These `lenient()`s seem redundant (as the corresponding methods are always called under this test), so they should be dropped.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/incoming/IncomingSnapshotCopier.java:
##########
@@ -226,18 +207,7 @@ private CompletableFuture<?> prepareTxStatePartitionStorageForRebalance(Executor
return completedFuture(null);
}
- return CompletableFuture.supplyAsync(() -> partitionSnapshotStorage.partition().reCreateTxStatePartitionStorage(), executor)
- .thenCompose(txStatePartitionStorage -> {
- if (canceled) {
- return completedFuture(null);
- }
-
- txStatePartitionStorage.lastApplied(FULL_RABALANCING_STARTED, 0);
-
- LOG.info("Copier prepared transaction state storage for the partition [partId={}, tableId={}]", partId(), tableId());
Review Comment:
Same thing: don't we want to log this anymore?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/incoming/IncomingSnapshotCopier.java:
##########
@@ -186,32 +186,13 @@ public SnapshotReader getReader() {
/**
* Prepares the {@link MvPartitionStorage} for a full rebalance.
- *
- * <p>Recreates {@link MvPartitionStorage} and sets the last applied index to {@link TableManager#FULL_RABALANCING_STARTED} so that
- * when the node is restarted, we can understand that the full rebalance has not completed, and we need to clean up the storage from
- * garbage.
*/
- private CompletableFuture<?> prepareMvPartitionStorageForRebalance(Executor executor) {
+ private CompletableFuture<?> prepareMvPartitionStorageForRebalance() {
if (canceled) {
return completedFuture(null);
}
- return partitionSnapshotStorage.partition().reCreateMvPartitionStorage()
- .thenComposeAsync(mvPartitionStorage -> {
- if (canceled) {
- return completedFuture(null);
- }
-
- mvPartitionStorage.runConsistently(() -> {
- mvPartitionStorage.lastApplied(FULL_RABALANCING_STARTED, 0);
-
- return null;
- });
-
- LOG.info("Copier prepared multi-versioned storage for the partition [partId={}, tableId={}]", partId(), tableId());
-
- return completedFuture(null);
- }, executor);
+ return partitionSnapshotStorage.partition().reCreateMvPartitionStorage();
Review Comment:
The previous version of the code logged when it prepared the MV storage. Here, the logging is removed. Was this made on purpose, or should we still log it?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionAccessImpl.java:
##########
@@ -57,48 +70,147 @@ public PartitionKey partitionKey() {
}
@Override
- public MvPartitionStorage mvPartitionStorage() {
- MvPartitionStorage mvPartition = mvTableStorage.getMvPartition(partId());
+ public CompletableFuture<Void> reCreateMvPartitionStorage() throws StorageException {
+ assert mvTableStorage.getMvPartition(partitionId()) != null : "table=" + tableName() + ", part=" + partitionId();
+
+ // TODO: IGNITE-18030 - actually recreate or do in a different way
+ //return mvTableStorage.destroyPartition(partId())
+ return CompletableFuture.completedFuture(null)
+ .thenApply(unused -> {
+ MvPartitionStorage mvPartitionStorage = mvTableStorage.getOrCreateMvPartition(partitionId());
+
+ mvPartitionStorage.runConsistently(() -> {
+ mvPartitionStorage.lastApplied(FULL_RABALANCING_STARTED, 0);
- assert mvPartition != null : "table=" + tableName() + ", part=" + partId();
+ return null;
+ });
- return mvPartition;
+ return null;
+ });
}
@Override
- public TxStateStorage txStatePartitionStorage() {
- TxStateStorage txStatePartitionStorage = txStateTableStorage.getTxStateStorage(partId());
+ public void reCreateTxStatePartitionStorage() throws StorageException {
+ assert txStateTableStorage.getTxStateStorage(partitionId()) != null : "table=" + tableName() + ", part=" + partitionId();
+
+ // TODO: IGNITE-18030 - actually recreate or do in a different way
+ //txStateTableStorage.destroyTxStateStorage(partId());
+
+ txStateTableStorage.getOrCreateTxStateStorage(partitionId()).lastApplied(FULL_RABALANCING_STARTED, 0);
+ }
- assert txStatePartitionStorage != null : "table=" + tableName() + ", part=" + partId();
+ private int partitionId() {
+ return partitionKey.partitionId();
+ }
- return txStatePartitionStorage;
+ private String tableName() {
+ return mvTableStorage.configuration().name().value();
}
@Override
- public CompletableFuture<MvPartitionStorage> reCreateMvPartitionStorage() throws StorageException {
- assert mvTableStorage.getMvPartition(partId()) != null : "table=" + tableName() + ", part=" + partId();
+ public Cursor<IgniteBiTuple<UUID, TxMeta>> getAllTxMeta() {
+ return getTxStateStorage(partitionId()).scan();
+ }
- // TODO: IGNITE-18030 - actually recreate or do in a different way
- //return mvTableStorage.destroyPartition(partId())
- return CompletableFuture.completedFuture(null)
- .thenApply(unused -> mvTableStorage.getOrCreateMvPartition(partId()));
+ @Override
+ public void addTxMeta(UUID txId, TxMeta txMeta) {
+ getTxStateStorage(partitionId()).put(txId, txMeta);
}
@Override
- public TxStateStorage reCreateTxStatePartitionStorage() throws StorageException {
- assert txStateTableStorage.getTxStateStorage(partId()) != null : "table=" + tableName() + ", part=" + partId();
+ public @Nullable RowId closestRowId(RowId lowerBound) {
+ return getMvPartitionStorage(partitionId()).closestRowId(lowerBound);
+ }
- // TODO: IGNITE-18030 - actually recreate or do in a different way
- //txStateTableStorage.destroyTxStateStorage(partId());
+ @Override
+ public Cursor<ReadResult> getAllRowVersions(RowId rowId) {
+ return getMvPartitionStorage(partitionId()).scanVersions(rowId);
+ }
- return txStateTableStorage.getOrCreateTxStateStorage(partId());
+ @Override
+ public @Nullable RaftGroupConfiguration committedGroupConfiguration() {
+ return getMvPartitionStorage(partitionId()).committedGroupConfiguration();
}
- private int partId() {
- return partitionKey.partitionId();
+ @Override
+ public void addWrite(RowId rowId, TableRow row, UUID txId, UUID commitTableId, int commitPartitionId) {
+ MvPartitionStorage mvPartitionStorage = getMvPartitionStorage(partitionId());
+
+ mvPartitionStorage.runConsistently(() -> mvPartitionStorage.addWrite(rowId, row, txId, commitTableId, commitPartitionId));
}
- private String tableName() {
- return mvTableStorage.configuration().name().value();
+ @Override
+ public void addWriteCommitted(RowId rowId, TableRow row, HybridTimestamp commitTimestamp) {
+ MvPartitionStorage mvPartitionStorage = getMvPartitionStorage(partitionId());
+
+ mvPartitionStorage.runConsistently(() -> {
+ mvPartitionStorage.addWriteCommitted(rowId, row, commitTimestamp);
+
+ return null;
+ });
+ }
+
+ @Override
+ public void updateLastApplied(long lastAppliedIndex, long lastAppliedTerm, RaftGroupConfiguration raftGroupConfig) {
+ MvPartitionStorage mvPartitionStorage = getMvPartitionStorage(partitionId());
+ TxStateStorage txStateStorage = getTxStateStorage(partitionId());
+
+ mvPartitionStorage.runConsistently(() -> {
+ mvPartitionStorage.lastApplied(lastAppliedIndex, lastAppliedTerm);
+
+ txStateStorage.lastApplied(lastAppliedIndex, lastAppliedTerm);
+
+ mvPartitionStorage.committedGroupConfiguration(raftGroupConfig);
+
+ return null;
+ });
+ }
+
+ @Override
+ public long minLastAppliedIndex() {
Review Comment:
Please add unit tests for all 4 min/max methods. This logic was implemented in other places before this change, and it was covered by tests (as it seems really important). So we probably need these tests here now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org