sashapolo commented on code in PR #3409:
URL: https://github.com/apache/ignite-3/pull/3409#discussion_r1524334669
##########
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java:
##########
@@ -820,57 +762,258 @@ private static <T> List<T> getAll(Cursor<T> cursor) {
}
}
- private void checkStorageDestroyed(MvPartitionStorage storage) {
- int partId = PARTITION_ID;
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ private static void checkCursorAfterStartRebalance(Cursor<?> cursor) {
+ assertDoesNotThrow(cursor::close);
- assertThrows(StorageClosedException.class, () ->
storage.runConsistently(locker -> null));
+ assertThrows(StorageRebalanceException.class, cursor::hasNext);
+ assertThrows(StorageRebalanceException.class, cursor::next);
- assertThrows(StorageClosedException.class, storage::flush);
+ if (cursor instanceof PeekCursor) {
+ assertThrows(StorageRebalanceException.class, ((PeekCursor<?>)
cursor)::peek);
+ }
+ }
- assertThrows(StorageClosedException.class, storage::lastAppliedIndex);
- assertThrows(StorageClosedException.class, storage::lastAppliedTerm);
- assertThrows(StorageClosedException.class,
storage::committedGroupConfiguration);
+ private static void fillStorages(
+ MvPartitionStorage mvPartitionStorage,
+ HashIndexStorage hashIndexStorage,
+ SortedIndexStorage sortedIndexStorage,
+ List<IgniteTuple3<RowId, BinaryRow, HybridTimestamp>> rows
+ ) {
+ assertThat(rows, hasSize(greaterThanOrEqualTo(2)));
- RowId rowId = new RowId(partId);
+ for (int i = 0; i < rows.size(); i++) {
+ int finalI = i;
- HybridTimestamp timestamp = clock.now();
+ IgniteTuple3<RowId, BinaryRow, HybridTimestamp> row = rows.get(i);
- assertThrows(StorageClosedException.class, () -> storage.read(new
RowId(PARTITION_ID), timestamp));
+ RowId rowId = row.get1();
+ BinaryRow binaryRow = row.get2();
+ HybridTimestamp timestamp = row.get3();
- BinaryRow binaryRow = binaryRow(new TestKey(0, "0"), new TestValue(1,
"1"));
+ assertNotNull(rowId);
+ assertNotNull(binaryRow);
+ assertNotNull(timestamp);
- assertThrows(StorageClosedException.class, () ->
storage.addWrite(rowId, binaryRow, UUID.randomUUID(), COMMIT_TABLE_ID, partId));
- assertThrows(StorageClosedException.class, () ->
storage.commitWrite(rowId, timestamp));
- assertThrows(StorageClosedException.class, () ->
storage.abortWrite(rowId));
- assertThrows(StorageClosedException.class, () ->
storage.addWriteCommitted(rowId, binaryRow, timestamp));
+ IndexRow hashIndexRow =
indexRow(hashIndexStorage.indexDescriptor(), binaryRow, rowId);
+ IndexRow sortedIndexRow =
indexRow(sortedIndexStorage.indexDescriptor(), binaryRow, rowId);
- assertThrows(StorageClosedException.class, () ->
storage.scan(timestamp));
- assertThrows(StorageClosedException.class, () ->
storage.scanVersions(rowId));
- assertThrows(StorageClosedException.class, () ->
storage.scanVersions(rowId));
+ mvPartitionStorage.runConsistently(locker -> {
+ locker.lock(rowId);
- assertThrows(StorageClosedException.class, () ->
storage.closestRowId(rowId));
+ if ((finalI % 2) == 0) {
+ mvPartitionStorage.addWrite(rowId, binaryRow,
UUID.randomUUID(), COMMIT_TABLE_ID, rowId.partitionId());
- assertThrows(StorageClosedException.class, storage::rowsCount);
- }
+ mvPartitionStorage.commitWrite(rowId, timestamp);
+ } else {
+ mvPartitionStorage.addWriteCommitted(rowId, binaryRow,
timestamp);
+ }
- private void checkStorageDestroyed(SortedIndexStorage storage) {
- checkStorageDestroyed((IndexStorage) storage);
+ hashIndexStorage.put(hashIndexRow);
- assertThrows(StorageClosedException.class, () -> storage.scan(null,
null, 0));
+ sortedIndexStorage.put(sortedIndexRow);
+
+ return null;
+ });
+ }
}
- private void checkStorageDestroyed(IndexStorage storage) {
- assertThrows(StorageClosedException.class, () ->
storage.get(mock(BinaryTuple.class)));
+ private static void checkForMissingRows(
+ MvPartitionStorage mvPartitionStorage,
+ HashIndexStorage hashIndexStorage,
+ SortedIndexStorage sortedIndexStorage,
+ List<IgniteTuple3<RowId, BinaryRow, HybridTimestamp>> rows
+ ) {
+ for (IgniteTuple3<RowId, BinaryRow, HybridTimestamp> row : rows) {
+ List<ReadResult> allVersions =
mvPartitionStorage.runConsistently(locker -> {
+ locker.lock(row.get1());
- assertThrows(StorageClosedException.class, () ->
storage.put(mock(IndexRow.class)));
+ return getAll(mvPartitionStorage.scanVersions(row.get1()));
+ });
+ assertThat(allVersions, is(empty()));
- assertThrows(StorageClosedException.class, () ->
storage.remove(mock(IndexRow.class)));
+ IndexRow hashIndexRow =
indexRow(hashIndexStorage.indexDescriptor(), row.get2(), row.get1());
+ IndexRow sortedIndexRow =
indexRow(sortedIndexStorage.indexDescriptor(), row.get2(), row.get1());
+
+
assertThat(getAll(hashIndexStorage.get(hashIndexRow.indexColumns())),
is(empty()));
+
assertThat(getAll(sortedIndexStorage.get(sortedIndexRow.indexColumns())),
is(empty()));
+ }
}
private int getPartitionIdOutOfRange() {
return tableStorage.getTableDescriptor().getPartitions();
}
+ private static void checkForPresenceRows(
+ MvPartitionStorage mvPartitionStorage,
+ HashIndexStorage hashIndexStorage,
+ SortedIndexStorage sortedIndexStorage,
+ List<IgniteTuple3<RowId, BinaryRow, HybridTimestamp>> rows
+ ) {
+ for (IgniteTuple3<RowId, BinaryRow, HybridTimestamp> row : rows) {
+ List<BinaryRow> allVersions =
mvPartitionStorage.runConsistently(locker -> {
+ locker.lock(row.get1());
+
+ return
toListOfBinaryRows(mvPartitionStorage.scanVersions(row.get1()));
+ });
+
+ assertThat(allVersions, contains(equalToRow(row.get2())));
+
+ IndexRow hashIndexRow =
indexRow(hashIndexStorage.indexDescriptor(), row.get2(), row.get1());
+ IndexRow sortedIndexRow =
indexRow(sortedIndexStorage.indexDescriptor(), row.get2(), row.get1());
+
+
assertThat(getAll(hashIndexStorage.get(hashIndexRow.indexColumns())),
contains(row.get1()));
+
assertThat(getAll(sortedIndexStorage.get(sortedIndexRow.indexColumns())),
contains(row.get1()));
+ }
+ }
+
+ /**
+ * Initializes the internal structures needed for tests.
+ *
+ * <p>This method *MUST* always be called in either subclass' constructor
or setUp method.
+ */
+ protected final void initialize() {
+ createTestTableAndIndexes(catalogService);
+
+ this.tableStorage = createMvTableStorage();
+
+ CatalogTableDescriptor catalogTableDescriptor =
catalogService.table(TABLE_NAME, clock.nowLong());
+ assertNotNull(catalogTableDescriptor);
+
+ CatalogIndexDescriptor catalogSortedIndexDescriptor =
catalogService.aliveIndex(SORTED_INDEX_NAME, clock.nowLong());
+ CatalogIndexDescriptor catalogHashIndexDescriptor =
catalogService.aliveIndex(HASH_INDEX_NAME, clock.nowLong());
+
+ assertNotNull(catalogSortedIndexDescriptor);
+ assertNotNull(catalogHashIndexDescriptor);
+
+ sortedIdx = new StorageSortedIndexDescriptor(catalogTableDescriptor,
(CatalogSortedIndexDescriptor) catalogSortedIndexDescriptor);
+ hashIdx = new StorageHashIndexDescriptor(catalogTableDescriptor,
(CatalogHashIndexDescriptor) catalogHashIndexDescriptor);
+ }
+
+ private static void
checkHashIndexStorageMethodsAfterStartRebalance(HashIndexStorage storage) {
+ assertDoesNotThrow(storage::indexDescriptor);
+
+ assertThrows(StorageRebalanceException.class, () ->
storage.get(mock(BinaryTuple.class)));
+ assertThrows(StorageRebalanceException.class, () ->
storage.remove(mock(IndexRow.class)));
+ }
+
+ @SuppressWarnings("resource")
+ @Test
+ public void testDestroyPartition() throws Exception {
+ assertThrows(IllegalArgumentException.class, () ->
tableStorage.destroyPartition(getPartitionIdOutOfRange()));
+
+ MvPartitionStorage mvPartitionStorage =
getOrCreateMvPartition(PARTITION_ID);
+ HashIndexStorage hashIndexStorage =
tableStorage.getOrCreateHashIndex(PARTITION_ID, hashIdx);
+ SortedIndexStorage sortedIndexStorage =
tableStorage.getOrCreateSortedIndex(PARTITION_ID, sortedIdx);
+
+ RowId rowId = new RowId(PARTITION_ID);
+
+ BinaryRow binaryRow = binaryRow(new TestKey(0, "0"), new TestValue(1,
"1"));
+
+ IndexRow hashIndexRow = indexRow(hashIndexStorage.indexDescriptor(),
binaryRow, rowId);
+ IndexRow sortedIndexRow =
indexRow(sortedIndexStorage.indexDescriptor(), binaryRow, rowId);
+
+ mvPartitionStorage.runConsistently(locker -> {
+ locker.lock(rowId);
+
+ mvPartitionStorage.addWriteCommitted(rowId, binaryRow,
clock.now());
+
+ hashIndexStorage.put(hashIndexRow);
+
+ sortedIndexStorage.put(sortedIndexRow);
+
+ return null;
+ });
+
+ PartitionTimestampCursor scanAtTimestampCursor =
mvPartitionStorage.scan(clock.now());
+ PartitionTimestampCursor scanLatestCursor =
mvPartitionStorage.scan(HybridTimestamp.MAX_VALUE);
+
+ Cursor<ReadResult> scanVersionsCursor =
mvPartitionStorage.runConsistently(locker -> {
+ locker.lock(rowId);
+
+ return mvPartitionStorage.scanVersions(rowId);
+ });
+
+ Cursor<RowId> getFromHashIndexCursor =
hashIndexStorage.get(hashIndexRow.indexColumns());
+
+ Cursor<RowId> getFromSortedIndexCursor =
sortedIndexStorage.get(hashIndexRow.indexColumns());
+ Cursor<IndexRow> scanFromSortedIndexCursor =
sortedIndexStorage.scan(null, null, GREATER);
+
+ assertThat(tableStorage.destroyPartition(PARTITION_ID),
willCompleteSuccessfully());
+
+ // Let's check that we won't get destroyed storages.
+ assertNull(tableStorage.getMvPartition(PARTITION_ID));
+ assertThrows(StorageException.class, () ->
tableStorage.getOrCreateHashIndex(PARTITION_ID, hashIdx));
+ assertThrows(StorageException.class, () ->
tableStorage.getOrCreateSortedIndex(PARTITION_ID, sortedIdx));
+
+ checkStorageDestroyed(mvPartitionStorage);
+ checkStorageDestroyed(hashIndexStorage);
+ checkStorageDestroyed(sortedIndexStorage);
+
+ assertThrows(StorageClosedException.class, () ->
getAll(scanAtTimestampCursor));
+ assertThrows(StorageClosedException.class, () ->
getAll(scanLatestCursor));
+
+ assertThrows(StorageClosedException.class, () ->
getAll(scanVersionsCursor));
+
+ assertThrows(StorageClosedException.class, () ->
getAll(getFromHashIndexCursor));
+
+ assertThrows(StorageClosedException.class, () ->
getAll(getFromSortedIndexCursor));
+ assertThrows(StorageClosedException.class, () ->
getAll(scanFromSortedIndexCursor));
+
+ // What happens if there is no partition?
+ assertThrows(StorageException.class, () ->
tableStorage.destroyPartition(PARTITION_ID));
+ }
+
+ @SuppressWarnings("resource")
+ @Test
+ public void testDestroyTableStorage() {
+ MvPartitionStorage mvPartitionStorage =
getOrCreateMvPartition(PARTITION_ID);
+ HashIndexStorage hashIndexStorage =
tableStorage.getOrCreateHashIndex(PARTITION_ID, hashIdx);
+ SortedIndexStorage sortedIndexStorage =
tableStorage.getOrCreateSortedIndex(PARTITION_ID, sortedIdx);
+
+ List<IgniteTuple3<RowId, BinaryRow, HybridTimestamp>> rows = List.of(
Review Comment:
I've added a `TestRow` class, please use it instead of the tuple
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
