You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by GitBox <gi...@apache.org> on 2021/02/08 12:02:54 UTC
[GitHub] [ignite] Mmuzaf opened a new pull request #8767: IGNITE-14134 commit snapshot partition iterator
Mmuzaf opened a new pull request #8767:
URL: https://github.com/apache/ignite/pull/8767
Thank you for submitting the pull request to the Apache Ignite.
In order to streamline the review of the contribution
we ask you to ensure the following steps have been taken:
### The Contribution Checklist
- [ ] There is a single JIRA ticket related to the pull request.
- [ ] The web-link to the pull request is attached to the JIRA ticket.
- [ ] The JIRA ticket has the _Patch Available_ state.
- [ ] The pull request body describes changes that have been made.
The description explains _WHAT_ and _WHY_ was made instead of _HOW_.
- [ ] The pull request title is treated as the final commit message.
The following pattern must be used: `IGNITE-XXXX Change summary` where `XXXX` - number of JIRA issue.
- [ ] A reviewer has been mentioned through the JIRA comments
(see [the Maintainers list](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute#HowtoContribute-ReviewProcessandMaintainers))
- [ ] The pull request has been checked by the Teamcity Bot and
the `green visa` attached to the JIRA ticket (see [TC.Bot: Check PR](https://mtcga.gridgain.com/prs.html))
### Notes
- [How to Contribute](https://cwiki.apache.org/confluence/display/IGNITE/How+to+Contribute)
- [Coding abbreviation rules](https://cwiki.apache.org/confluence/display/IGNITE/Abbreviation+Rules)
- [Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Coding+Guidelines)
- [Apache Ignite Teamcity Bot](https://cwiki.apache.org/confluence/display/IGNITE/Apache+Ignite+Teamcity+Bot)
If you need any help, please email dev@ignite.apache.org or ask anу advice on http://asf.slack.com _#ignite_ channel.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591815051
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,58 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param folderName The node folder name, usually it's the same as the U.maskForFileName(consistentId).
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String folderName,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(folderName));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ List<File> grps = cacheDirectories(nodePath, name -> name.equals(grpName));
+
+ if (F.isEmpty(grps) || grps.size() > 1)
+ throw new IgniteCheckedException("Snapshot cache group not found [dir=" + snpDir.getAbsolutePath() + ", grpName=" + grpName + ']');
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591777005
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
+ int keysPerCache = 3700;
+ int cacheValSize = 33;
+ int rowsOnPage = 37;
+
+ CacheConfiguration<Integer, Value> ccfg1 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+ CacheConfiguration<Integer, Value> ccfg2 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx2"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ ccfg1.setGroupName(DEFAULT_CACHE_NAME);
+ ccfg2.setGroupName(DEFAULT_CACHE_NAME);
+
+ IgniteEx ignite = startGridsWithCache(1, keysPerCache, k -> new Value(new byte[cacheValSize]), ccfg1, ccfg2);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ AtomicReference<ByteBuffer> buffRef = new AtomicReference<>();
+ AtomicReference<Long> pageIdRef = new AtomicReference<>();
+
+ snapshotStoreFactory(ignite, new BiFunction<Integer, Boolean, FileVersionCheckingFactory>() {
+ @Override public FileVersionCheckingFactory apply(Integer grpId, Boolean encrypted) {
+ FilePageStoreManager storeMgr = (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ return new FileVersionCheckingFactory(storeMgr.getPageStoreFileIoFactory(), storeMgr.getPageStoreFileIoFactory(), storeMgr::pageSize) {
+ @Override public PageStore createPageStore(
+ byte type,
+ IgniteOutClosure<Path> pathProvider,
+ LongConsumer allocatedTracker
+ ) {
+ return new FilePageStoreV2(
+ type,
+ pathProvider,
+ storeMgr.getPageStoreFileIoFactory(),
+ storeMgr.pageSize(),
+ allocatedTracker
+ ) {
+ @Override public boolean read(
+ long pageId,
+ ByteBuffer pageBuf,
+ boolean keepCrc
+ ) throws IgniteCheckedException {
+ buffRef.set(pageBuf);
+ pageIdRef.set(pageId);
+
+ return super.read(pageId, pageBuf, keepCrc);
+ }
+ };
+ }
+ };
+ }
+ });
+
+ int currRows = 0;
+ int rowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r593042658
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -317,4 +363,175 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckFailsOnPartitionDataDiffers() throws Exception {
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ ignite.getOrCreateCache(ccfg).put(1, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(PART_ID);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ // Change the cache value only at on of the cluster node to get hash conflict when the check command ends.
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(PART_ID);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = new BinaryObjectImpl(binCtx, binCtx.marshaller().marshal(new Value(bytes)), 0);
+
+ boolean success = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(PART_ID);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(PART_ID));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(),
+ "The check procedure has finished, found 1 conflict partitions: [counterConflicts=0, hashConflicts=1]");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckHashesSameAsIdleVerifyHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleVerifyHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpCheckHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
+
+ assertNotNull(idleVerifyHashes);
+ assertNotNull(snpCheckHashes);
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591766440
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
Review comment:
Fixed.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
+
+ /**
+ * @param hashes Map of calculated partition hashes.
+ */
+ public TestVisorBackupPartitionsTask(Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes) {
+ this.hashes = hashes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable IdleVerifyResultV2 reduce(List<ComputeJobResult> results) throws IgniteException {
+ IdleVerifyResultV2 res = super.reduce(results);
+
+ for (ComputeJobResult job : results) {
+ if (job.getException() != null)
+ continue;
+
+ job.<Map<PartitionKeyV2, PartitionHashRecordV2>>getData().forEach((k, v) ->
+ hashes.computeIfAbsent(k, k0 -> new ArrayList<>()).add(v));
+ }
+
+ return res;
+ }
+ }
+
+ /** */
+ private static class TestSnapshotPartitionsVerifyTask extends SnapshotPartitionsVerifyTask {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591748350
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
+ private final BitSet markedPages;
+
+ /** Pages which already read and must be skipped. */
+ private final BitSet readPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration though partition reached its end. */
+ private boolean secondScanComplete;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx;
+
+ /**
+ * During scanning a cache partition presented as {@code PageStore} we must guarantee the following:
+ * all the pages of this storage remains unchanged during the Iterator remains opened, the stored data
+ * keeps its consistency. We can't read the {@code PageStore} during an ongoing checkpoint over it.
+ *
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.ensure();
+ pages = store.pages();
+ markedPages = new BitSet(pages);
+ readPages = new BitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + readPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ return false;
+
+ try {
+ for (; currIdx < 2 * pages && rows.isEmpty(); currIdx++) {
+ boolean first = currIdx < pages;
+ int pageIdx = currIdx % pages;
+
+ if (readPages.get(pageIdx) || (!first && markedPages.get(pageIdx)))
+ continue;
+
+ if (!readPageFromStore(pageId(partId, FLAG_DATA, pageIdx), locBuff)) {
+ // Skip not FLAG_DATA pages.
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = bufferAddress(locBuff);
+ DataPageIO io = getPageIO(T_DATA, getVersion(pageAddr));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (first) {
+ // Skip empty pages.
+ if (rowsCnt == 0) {
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ // There is no difference between a page containing an incomplete DataRow fragment and
+ // the page where DataRow takes up all the free space. There is no a dedicated
+ // flag for this case in page header.
+ // During the storage scan we can skip such pages at the first iteration over the partition file,
+ // since all the fragmented pages will be marked by BitSet array we will safely read the others
+ // on the second iteration.
+ if (freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ changeBit(markedPages, pageIndex(pageId(link)));
+
+ continue;
+ }
+ }
+
+ changeBit(readPages, pageIdx);
+
+ for (int itemId = 0; itemId < rowsCnt; itemId++) {
+ DataRow row = new DataRow();
+
+ row.partition(partId);
+
+ row.initFromPageBuffer(
+ sctx,
+ coctx,
+ new IgniteThrowableFunction<Long, ByteBuffer>() {
+ @Override public ByteBuffer apply(Long nextPageId) throws IgniteCheckedException {
+ boolean success = readPageFromStore(nextPageId, fragmentBuff);
+
+ assert success : "Only FLAG_DATA pages allowed: " + toDetailString(nextPageId);
+
+ // Fragment of page has been read, might be skipped further.
+ changeBit(readPages, pageIndex(nextPageId));
+
+ return fragmentBuff;
+ }
+ },
+ locBuff,
+ io,
+ itemId,
+ false,
+ CacheDataRowAdapter.RowData.FULL,
+ false);
+
+ rows.add(row);
+ }
+ }
+
+ if (currIdx == 2 * pages) {
+ secondScanComplete = true;
+
+ boolean set = true;
+
+ for (int j = 0; j < pages; j++)
+ set &= readPages.get(j);
+
+ assert set : "readPages=" + readPages + ", pages=" + pages;
+ }
+
+ return !rows.isEmpty();
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteCheckedException("Error during iteration through page store: " + this, e);
+ }
+ }
+
+ /**
+ * @param bitSet BitSet to change bit index.
+ * @param idx Index of bit to change.
+ */
+ private static void changeBit(BitSet bitSet, int idx) {
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
+ private final BitSet markedPages;
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1142,58 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param folderName The node folder name, usually it's the same as the U.maskForFileName(consistentId).
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584881518
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
Review comment:
I've rollbacked the changes related to the AtomicBitSet. `BitSet` used insted.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591753829
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
##########
@@ -504,105 +500,63 @@ private boolean isCacheMatchFilter(DynamicCacheDescriptor desc) {
}
/**
- * @param grpCtx Group context.
+ * @param gctx Group context.
* @param part Local partition.
*/
private Future<Map<PartitionKeyV2, PartitionHashRecordV2>> calculatePartitionHashAsync(
Review comment:
Actually, it will require additional changes which I've tried to avoid during the implementation. It would be safe to perform such changes in a separate task. WDYT?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584881974
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
Review comment:
I've updated the comment, please, take a look.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591765596
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591758692
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584880687
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
Review comment:
Fixed. The `ensure` method is used in such cases through other parts of the source code.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(U.maskForFileName(consId)));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ File[] grps = nodePath.listFiles(path -> path.isDirectory() &&
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(U.maskForFileName(consId)));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ File[] grps = nodePath.listFiles(path -> path.isDirectory() &&
+ (path.getName().equalsIgnoreCase(cacheDirName(true, grpName)) ||
+ path.getName().equalsIgnoreCase(cacheDirName(false, grpName))));
Review comment:
Agree, fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591764311
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591776895
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
+ int keysPerCache = 3700;
+ int cacheValSize = 33;
+ int rowsOnPage = 37;
+
+ CacheConfiguration<Integer, Value> ccfg1 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+ CacheConfiguration<Integer, Value> ccfg2 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx2"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ ccfg1.setGroupName(DEFAULT_CACHE_NAME);
+ ccfg2.setGroupName(DEFAULT_CACHE_NAME);
+
+ IgniteEx ignite = startGridsWithCache(1, keysPerCache, k -> new Value(new byte[cacheValSize]), ccfg1, ccfg2);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ AtomicReference<ByteBuffer> buffRef = new AtomicReference<>();
+ AtomicReference<Long> pageIdRef = new AtomicReference<>();
+
+ snapshotStoreFactory(ignite, new BiFunction<Integer, Boolean, FileVersionCheckingFactory>() {
+ @Override public FileVersionCheckingFactory apply(Integer grpId, Boolean encrypted) {
+ FilePageStoreManager storeMgr = (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ return new FileVersionCheckingFactory(storeMgr.getPageStoreFileIoFactory(), storeMgr.getPageStoreFileIoFactory(), storeMgr::pageSize) {
+ @Override public PageStore createPageStore(
+ byte type,
+ IgniteOutClosure<Path> pathProvider,
+ LongConsumer allocatedTracker
+ ) {
+ return new FilePageStoreV2(
+ type,
+ pathProvider,
+ storeMgr.getPageStoreFileIoFactory(),
+ storeMgr.pageSize(),
+ allocatedTracker
+ ) {
+ @Override public boolean read(
+ long pageId,
+ ByteBuffer pageBuf,
+ boolean keepCrc
+ ) throws IgniteCheckedException {
+ buffRef.set(pageBuf);
+ pageIdRef.set(pageId);
+
+ return super.read(pageId, pageBuf, keepCrc);
+ }
+ };
+ }
+ };
+ }
+ });
+
+ int currRows = 0;
+ int rowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(buffRef.get()));
+
+ long pageAddr = GridUnsafe.bufferAddress(buffRef.get());
+ rowsCnt = io.getDirectCount(pageAddr);
+
+ assertEquals(cacheValSize, ((Value)row.value().value(coctx, false)).arr().length);
+ assertEquals(rowsOnPage, rowsCnt);
+
+ currRows++;
+ }
+ }
+
+ assertEquals(keysPerCache, currRows / 2);
+
+ // Remove all rows in the middle of the page.
+ for (int k = 0; k < keysPerCache; k++) {
+ if (k % rowsOnPage == 1 || k % rowsOnPage == (rowsOnPage - 1))
+ continue;
+
+ ignite.getOrCreateCache(ccfg1).remove(k);
+ ignite.getOrCreateCache(ccfg2).remove(k);
+ }
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME + 2).get();
+
+ int indRowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME + 2,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591766308
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591770980
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
+
+ /**
+ * @param hashes Map of calculated partition hashes.
+ */
+ public TestVisorBackupPartitionsTask(Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes) {
Review comment:
I doubt we need it here, since the amount of changes may only be greater.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584882187
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+
+import java.util.concurrent.atomic.AtomicIntegerArray;
+
+/**
+ * Atomic bitset array.
+ */
+class AtomicBitSet {
+ /**
+ * Container of bits.
+ */
+ private final AtomicIntegerArray arr;
+
+ /**
+ * Size of array of bits.
+ */
+ private final int size;
+
+ /**
+ * @param size Size of array.
+ */
+ public AtomicBitSet(int size) {
+ this.size = size;
+
+ arr = new AtomicIntegerArray((size + 31) >>> 5);
+ }
+
+ /**
+ * @return Size of bitset in bits.
+ */
+ public int size() {
+ return size;
+ }
+
+ /**
+ * @param idx Index in bitset array.
+ * @return {@code true} if the bit is set.
+ */
+ public boolean check(long idx) {
+ if (idx >= size)
+ return false;
+
+ int bit = 1 << idx;
+ int bucket = (int)(idx >>> 5);
+
+ int cur = arr.get(bucket);
+
+ return (cur & bit) == bit;
+ }
+
+ /**
+ * @param off Bit position to change.
+ * @return {@code true} if bit has been set,
+ * {@code false} if bit changed by another thread or out of range.
+ */
+ public boolean touch(long off) {
Review comment:
I've rollbacked the changes related to the AtomicBitSet. BitSet used instead.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
##########
@@ -223,6 +228,66 @@ public static String formatUpdateCountersDiff(IgniteEx ig, List<Integer> diff) {
return diff;
}
+ /**
+ * @param updCntr Partition update counter prior check.
+ * @param grpId Group id.
+ * @param partId Partition id.
+ * @param grpName Group name.
+ * @param consId Local node consistent id.
+ * @param state Partition state to check.
+ * @param isPrimary {@code true} if partition is primary.
+ * @param partSize Partition size on disk.
+ * @param it Iterator though partition data rows.
+ * @throws IgniteCheckedException If fails.
+ * @return Map of calculated partition.
+ */
+ public static Map<PartitionKeyV2, PartitionHashRecordV2> calculatePartitionHash(
+ long updCntr,
+ int grpId,
+ int partId,
+ String grpName,
+ Object consId,
+ GridDhtPartitionState state,
+ boolean isPrimary,
+ long partSize,
+ GridIterator<CacheDataRow> it
+ ) throws IgniteCheckedException {
+ int partHash = 0;
+
+ PartitionKeyV2 partKey = new PartitionKeyV2(grpId, partId, grpName);
+
+ if (state == GridDhtPartitionState.MOVING || state == GridDhtPartitionState.LOST) {
+ PartitionHashRecordV2 movingHashRecord = new PartitionHashRecordV2(
+ partKey,
+ isPrimary,
+ consId,
+ partHash,
+ updCntr,
+ state == GridDhtPartitionState.MOVING ? PartitionHashRecordV2.MOVING_PARTITION_SIZE : 0,
+ state == GridDhtPartitionState.MOVING ? PartitionHashRecordV2.PartitionState.MOVING : PartitionHashRecordV2.PartitionState.LOST
+ );
+
+ return Collections.singletonMap(partKey, movingHashRecord);
+ }
+
+ if (state != GridDhtPartitionState.OWNING)
+ return emptyMap();
+
+ while (it.hasNextX()) {
+ CacheDataRow row = it.nextX();
+
+ partHash += row.key().hashCode();
+
+ // Object context may be null since the value bytes have been read directly from page.
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf merged pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf merged pull request #8767:
URL: https://github.com/apache/ignite/pull/8767
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
anton-vinogradov commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r585330088
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,58 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param folderName The node folder name, usually it's the same as the U.maskForFileName(consistentId).
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String folderName,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(folderName));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ List<File> grps = cacheDirectories(nodePath, name -> name.equals(grpName));
+
+ if (F.isEmpty(grps) || grps.size() > 1)
+ throw new IgniteCheckedException("Snapshot cache group not found [dir=" + snpDir.getAbsolutePath() + ", grpName=" + grpName + ']');
Review comment:
`Snapshot cache group not found` is an incorrect statement when `grps.size() > 1`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
Review comment:
Tail pages, contain only part of the value, to be skipped during the second iterations (will be processed via headers processing).
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
+ private final BitSet markedPages;
+
+ /** Pages which already read and must be skipped. */
+ private final BitSet readPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration though partition reached its end. */
+ private boolean secondScanComplete;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx;
+
+ /**
+ * During scanning a cache partition presented as {@code PageStore} we must guarantee the following:
+ * all the pages of this storage remains unchanged during the Iterator remains opened, the stored data
+ * keeps its consistency. We can't read the {@code PageStore} during an ongoing checkpoint over it.
+ *
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.ensure();
+ pages = store.pages();
+ markedPages = new BitSet(pages);
+ readPages = new BitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + readPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ return false;
+
+ try {
+ for (; currIdx < 2 * pages && rows.isEmpty(); currIdx++) {
+ boolean first = currIdx < pages;
+ int pageIdx = currIdx % pages;
+
+ if (readPages.get(pageIdx) || (!first && markedPages.get(pageIdx)))
+ continue;
+
+ if (!readPageFromStore(pageId(partId, FLAG_DATA, pageIdx), locBuff)) {
+ // Skip not FLAG_DATA pages.
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = bufferAddress(locBuff);
+ DataPageIO io = getPageIO(T_DATA, getVersion(pageAddr));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (first) {
+ // Skip empty pages.
+ if (rowsCnt == 0) {
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ // There is no difference between a page containing an incomplete DataRow fragment and
+ // the page where DataRow takes up all the free space. There is no a dedicated
+ // flag for this case in page header.
+ // During the storage scan we can skip such pages at the first iteration over the partition file,
+ // since all the fragmented pages will be marked by BitSet array we will safely read the others
+ // on the second iteration.
+ if (freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ changeBit(markedPages, pageIndex(pageId(link)));
+
+ continue;
+ }
+ }
+
+ changeBit(readPages, pageIdx);
+
+ for (int itemId = 0; itemId < rowsCnt; itemId++) {
+ DataRow row = new DataRow();
+
+ row.partition(partId);
+
+ row.initFromPageBuffer(
+ sctx,
+ coctx,
+ new IgniteThrowableFunction<Long, ByteBuffer>() {
+ @Override public ByteBuffer apply(Long nextPageId) throws IgniteCheckedException {
+ boolean success = readPageFromStore(nextPageId, fragmentBuff);
+
+ assert success : "Only FLAG_DATA pages allowed: " + toDetailString(nextPageId);
+
+ // Fragment of page has been read, might be skipped further.
+ changeBit(readPages, pageIndex(nextPageId));
+
+ return fragmentBuff;
+ }
+ },
+ locBuff,
+ io,
+ itemId,
+ false,
+ CacheDataRowAdapter.RowData.FULL,
+ false);
+
+ rows.add(row);
+ }
+ }
+
+ if (currIdx == 2 * pages) {
+ secondScanComplete = true;
+
+ boolean set = true;
+
+ for (int j = 0; j < pages; j++)
+ set &= readPages.get(j);
+
+ assert set : "readPages=" + readPages + ", pages=" + pages;
+ }
+
+ return !rows.isEmpty();
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteCheckedException("Error during iteration through page store: " + this, e);
+ }
+ }
+
+ /**
+ * @param bitSet BitSet to change bit index.
+ * @param idx Index of bit to change.
+ */
+ private static void changeBit(BitSet bitSet, int idx) {
Review comment:
Since we "changing", we should pass the value?
Another case is to replace `change` with `set`.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
Review comment:
Could we minimize `try` section?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
Review comment:
Do we really need a loop here?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
Review comment:
javadoc missed
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1142,58 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param folderName The node folder name, usually it's the same as the U.maskForFileName(consistentId).
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
Review comment:
Not sure, but can we have *iterator name for this method?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
Review comment:
Should we use maxKey instead of 15_000?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
+ int keysPerCache = 3700;
+ int cacheValSize = 33;
+ int rowsOnPage = 37;
+
+ CacheConfiguration<Integer, Value> ccfg1 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+ CacheConfiguration<Integer, Value> ccfg2 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx2"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ ccfg1.setGroupName(DEFAULT_CACHE_NAME);
+ ccfg2.setGroupName(DEFAULT_CACHE_NAME);
+
+ IgniteEx ignite = startGridsWithCache(1, keysPerCache, k -> new Value(new byte[cacheValSize]), ccfg1, ccfg2);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ AtomicReference<ByteBuffer> buffRef = new AtomicReference<>();
+ AtomicReference<Long> pageIdRef = new AtomicReference<>();
+
+ snapshotStoreFactory(ignite, new BiFunction<Integer, Boolean, FileVersionCheckingFactory>() {
+ @Override public FileVersionCheckingFactory apply(Integer grpId, Boolean encrypted) {
+ FilePageStoreManager storeMgr = (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ return new FileVersionCheckingFactory(storeMgr.getPageStoreFileIoFactory(), storeMgr.getPageStoreFileIoFactory(), storeMgr::pageSize) {
+ @Override public PageStore createPageStore(
+ byte type,
+ IgniteOutClosure<Path> pathProvider,
+ LongConsumer allocatedTracker
+ ) {
+ return new FilePageStoreV2(
+ type,
+ pathProvider,
+ storeMgr.getPageStoreFileIoFactory(),
+ storeMgr.pageSize(),
+ allocatedTracker
+ ) {
+ @Override public boolean read(
+ long pageId,
+ ByteBuffer pageBuf,
+ boolean keepCrc
+ ) throws IgniteCheckedException {
+ buffRef.set(pageBuf);
+ pageIdRef.set(pageId);
+
+ return super.read(pageId, pageBuf, keepCrc);
+ }
+ };
+ }
+ };
+ }
+ });
+
+ int currRows = 0;
+ int rowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
Review comment:
tx1 duplicate
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
Review comment:
As decided, this test checks internals already checked by values read on recovery.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
+ private final BitSet markedPages;
Review comment:
Let's rename them to tailPages or something like this.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
##########
@@ -223,6 +228,66 @@ public static String formatUpdateCountersDiff(IgniteEx ig, List<Integer> diff) {
return diff;
}
+ /**
+ * @param updCntr Partition update counter prior check.
+ * @param grpId Group id.
+ * @param partId Partition id.
+ * @param grpName Group name.
+ * @param consId Local node consistent id.
+ * @param state Partition state to check.
+ * @param isPrimary {@code true} if partition is primary.
+ * @param partSize Partition size on disk.
+ * @param it Iterator though partition data rows.
+ * @throws IgniteCheckedException If fails.
+ * @return Map of calculated partition.
+ */
+ public static Map<PartitionKeyV2, PartitionHashRecordV2> calculatePartitionHash(
Review comment:
Always returns singletonMap, is there any reason to return the map in this case?
Also, because of the method name, we expect not a map, but something related to the partition.
Also, this looks odd when we have the same (partId, grpId) at the input and at the output as well.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
+ ByteBuffer fragmentBuff = reader.apply(pageId);
+
+ long fragmentAddr = GridUnsafe.bufferAddress(fragmentBuff);
+ DataPageIO io2 = PageIO.getPageIO(T_DATA, PageIO.getVersion(fragmentBuff));
+
+ incomplete = readIncomplete(incomplete, sctx, coctx, fragmentBuff.capacity(), fragmentBuff.capacity(),
+ fragmentAddr, itemId(nextLink), io2, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ nextLink = incomplete.getNextLink();
+ }
+ catch (Exception e) {
+ throw new IgniteException("Error during reading DataRow [pageId=" + pageId + ']', e);
+ }
+ }
+ while (nextLink != 0);
+
+ assert isReady() : "ready";
Review comment:
String after the assert should represent a "fail description".
`ready` seems to be a "good case" :)
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
##########
@@ -504,105 +500,63 @@ private boolean isCacheMatchFilter(DynamicCacheDescriptor desc) {
}
/**
- * @param grpCtx Group context.
+ * @param gctx Group context.
* @param part Local partition.
*/
private Future<Map<PartitionKeyV2, PartitionHashRecordV2>> calculatePartitionHashAsync(
Review comment:
should we replace Map with Tuple here?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
Review comment:
Let's provide a comment that explains that we're changing the value at one of the nodes to get a hash check issue.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/binary/BinaryFieldExtractionSelfTest.java
##########
@@ -276,7 +276,7 @@ public void testDecimalFieldMarshalling() throws Exception {
* @param marsh Binary marshaller.
* @return Binary object.
*/
- protected BinaryObjectImpl toBinary(Object obj, BinaryMarshaller marsh) throws Exception {
+ public static BinaryObjectImpl toBinary(Object obj, BinaryMarshaller marsh) throws Exception {
Review comment:
Seems to be not a proper solution to make this static to use at different test.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
Review comment:
Do we have another `success` to call this `*0`?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
Review comment:
Should we check that hash verification fail is the failure reason?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
Review comment:
Let's think about naming. Now it tells that iterator will scan the page.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
+ ByteBuffer fragmentBuff = reader.apply(pageId);
+
+ long fragmentAddr = GridUnsafe.bufferAddress(fragmentBuff);
+ DataPageIO io2 = PageIO.getPageIO(T_DATA, PageIO.getVersion(fragmentBuff));
+
+ incomplete = readIncomplete(incomplete, sctx, coctx, fragmentBuff.capacity(), fragmentBuff.capacity(),
+ fragmentAddr, itemId(nextLink), io2, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ nextLink = incomplete.getNextLink();
+ }
+ catch (Exception e) {
Review comment:
Coul we set strict type here, eg. IgniteCheckedException thrown by `readIncomplete`?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
Review comment:
Let's replace 0 by constant referencing to partition id at the whole test.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
Review comment:
According to its name, this test should check hashes misses, but it checks hashes validation fail on different content.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
Review comment:
testClusterSnapshotHashesTheSameAsIdleVerifyHashes? %)
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
+
+ /**
+ * @param hashes Map of calculated partition hashes.
+ */
+ public TestVisorBackupPartitionsTask(Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes) {
Review comment:
Looks like TestSnapshotPartitionsVerifyTask code equals this code.
Can we resolve the duplication issue?
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
+
+ /**
+ * @param hashes Map of calculated partition hashes.
+ */
+ public TestVisorBackupPartitionsTask(Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes) {
+ this.hashes = hashes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable IdleVerifyResultV2 reduce(List<ComputeJobResult> results) throws IgniteException {
+ IdleVerifyResultV2 res = super.reduce(results);
+
+ for (ComputeJobResult job : results) {
+ if (job.getException() != null)
+ continue;
+
+ job.<Map<PartitionKeyV2, PartitionHashRecordV2>>getData().forEach((k, v) ->
+ hashes.computeIfAbsent(k, k0 -> new ArrayList<>()).add(v));
+ }
+
+ return res;
+ }
+ }
+
+ /** */
+ private static class TestSnapshotPartitionsVerifyTask extends SnapshotPartitionsVerifyTask {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
Review comment:
javadoc missed
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
+ int keysPerCache = 3700;
+ int cacheValSize = 33;
+ int rowsOnPage = 37;
+
+ CacheConfiguration<Integer, Value> ccfg1 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+ CacheConfiguration<Integer, Value> ccfg2 = txCacheConfig(new CacheConfiguration<Integer, Value>("tx2"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ ccfg1.setGroupName(DEFAULT_CACHE_NAME);
+ ccfg2.setGroupName(DEFAULT_CACHE_NAME);
+
+ IgniteEx ignite = startGridsWithCache(1, keysPerCache, k -> new Value(new byte[cacheValSize]), ccfg1, ccfg2);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ AtomicReference<ByteBuffer> buffRef = new AtomicReference<>();
+ AtomicReference<Long> pageIdRef = new AtomicReference<>();
+
+ snapshotStoreFactory(ignite, new BiFunction<Integer, Boolean, FileVersionCheckingFactory>() {
+ @Override public FileVersionCheckingFactory apply(Integer grpId, Boolean encrypted) {
+ FilePageStoreManager storeMgr = (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ return new FileVersionCheckingFactory(storeMgr.getPageStoreFileIoFactory(), storeMgr.getPageStoreFileIoFactory(), storeMgr::pageSize) {
+ @Override public PageStore createPageStore(
+ byte type,
+ IgniteOutClosure<Path> pathProvider,
+ LongConsumer allocatedTracker
+ ) {
+ return new FilePageStoreV2(
+ type,
+ pathProvider,
+ storeMgr.getPageStoreFileIoFactory(),
+ storeMgr.pageSize(),
+ allocatedTracker
+ ) {
+ @Override public boolean read(
+ long pageId,
+ ByteBuffer pageBuf,
+ boolean keepCrc
+ ) throws IgniteCheckedException {
+ buffRef.set(pageBuf);
+ pageIdRef.set(pageId);
+
+ return super.read(pageId, pageBuf, keepCrc);
+ }
+ };
+ }
+ };
+ }
+ });
+
+ int currRows = 0;
+ int rowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(buffRef.get()));
+
+ long pageAddr = GridUnsafe.bufferAddress(buffRef.get());
+ rowsCnt = io.getDirectCount(pageAddr);
+
+ assertEquals(cacheValSize, ((Value)row.value().value(coctx, false)).arr().length);
+ assertEquals(rowsOnPage, rowsCnt);
+
+ currRows++;
+ }
+ }
+
+ assertEquals(keysPerCache, currRows / 2);
+
+ // Remove all rows in the middle of the page.
+ for (int k = 0; k < keysPerCache; k++) {
+ if (k % rowsOnPage == 1 || k % rowsOnPage == (rowsOnPage - 1))
+ continue;
+
+ ignite.getOrCreateCache(ccfg1).remove(k);
+ ignite.getOrCreateCache(ccfg2).remove(k);
+ }
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME + 2).get();
+
+ int indRowsCnt;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME + 2,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex("tx1").context().cacheObjectContext();
Review comment:
tx1 duplicate
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
Review comment:
Should be const.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
Review comment:
12008 should be const with an explanation of why.
Should be calculated using page size, since default may be changed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584881779
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
Review comment:
Comment added.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591750597
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
anton-vinogradov commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r593019812
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -317,4 +363,175 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckFailsOnPartitionDataDiffers() throws Exception {
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ ignite.getOrCreateCache(ccfg).put(1, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(PART_ID);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ // Change the cache value only at on of the cluster node to get hash conflict when the check command ends.
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(PART_ID);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = new BinaryObjectImpl(binCtx, binCtx.marshaller().marshal(new Value(bytes)), 0);
+
+ boolean success = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(PART_ID);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(PART_ID));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(),
+ "The check procedure has finished, found 1 conflict partitions: [counterConflicts=0, hashConflicts=1]");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckHashesSameAsIdleVerifyHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleVerifyHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpCheckHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
+
+ assertNotNull(idleVerifyHashes);
+ assertNotNull(snpCheckHashes);
Review comment:
Let's assert they're not empty as well.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591769909
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591770980
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCompareHashes() throws Exception {
+ Random rnd = new Random();
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<>(DEFAULT_CACHE_NAME));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[rnd.nextInt(32768)]), ccfg);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> idleHashes = new HashMap<>();
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> snpHashes = new HashMap<>();
+
+ IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(idleHashes),
+ new VisorIdleVerifyTaskArg(new HashSet<>(Collections.singletonList(ccfg.getName())),
+ new HashSet<>(),
+ false,
+ CacheFilterEnum.USER,
+ true));
+
+ IdleVerifyResultV2 snpVerifyRes = ignite.compute().execute(new TestSnapshotPartitionsVerifyTask(snpHashes),
+ Collections.singletonMap(ignite.cluster().localNode(),
+ Collections.singletonList(snp(ignite).readSnapshotMetadata(SNAPSHOT_NAME, (String)ignite.configuration().getConsistentId()))));
+
+ assertEquals(idleHashes, snpHashes);
+ assertEquals(idleVerifyRes, snpVerifyRes);
+ }
+
+ /** */
+ private static class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes;
+
+ /**
+ * @param hashes Map of calculated partition hashes.
+ */
+ public TestVisorBackupPartitionsTask(Map<PartitionKeyV2, List<PartitionHashRecordV2>> hashes) {
Review comment:
I doubt we need it here, since the amount of changes may only increase.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584881619
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+
+import java.util.concurrent.atomic.AtomicIntegerArray;
+
+/**
+ * Atomic bitset array.
+ */
+class AtomicBitSet {
Review comment:
I've rollbacked the changes related to the AtomicBitSet. BitSet used instead.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591758630
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591762895
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
+ return;
+
+ GridIterator<CacheDataRow> it = cache.context().offheap().partitionIterator(0);
+
+ assertTrue(it.hasNext());
+
+ CacheDataRow row0 = it.nextX();
+
+ AffinityTopologyVersion topVer = cctx.exchange().readyAffinityVersion();
+ GridCacheEntryEx cached = cache.entryEx(row0.key(), topVer);
+
+ byte[] bytes = new byte[2000];
+ new Random().nextBytes(bytes);
+
+ try {
+ BinaryObjectImpl newVal = toBinary(new Value(bytes), binCtx.marshaller());
+
+ boolean success0 = cached.initialValue(
+ newVal,
+ new GridCacheVersion(row0.version().topologyVersion(),
+ row0.version().nodeOrder(),
+ row0.version().order() + 1),
+ null,
+ null,
+ TxState.NA,
+ TxState.NA,
+ TTL_ETERNAL,
+ row0.expireTime(),
+ true,
+ topVer,
+ DR_NONE,
+ false,
+ null);
+
+ assertTrue(success0);
+
+ long newPartCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+
+ assertEquals(newPartCtr, partCtr);
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+ }
+
+ @Override public void onCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+
+ @Override public void beforeCheckpointBegin(Context ctx) throws IgniteCheckedException {
+
+ }
+ });
+
+ db.waitForCheckpoint("test-checkpoint");
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
Review comment:
Fixed.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[2000]));
+
+ forceCheckpoint(ignite);
+
+ GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
+ GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)cctx.database();
+
+ BinaryContext binCtx = ((CacheObjectBinaryProcessorImpl)ignite.context().cacheObjects()).binaryContext();
+
+ GridCacheAdapter<?, ?> cache = ignite.context().cache().internalCache(dfltCacheCfg.getName());
+ long partCtr = cache.context().offheap().lastUpdatedPartitionCounter(0);
+ AtomicBoolean done = new AtomicBoolean();
+
+ db.addCheckpointListener(new CheckpointListener() {
+ @Override public void onMarkCheckpointBegin(Context ctx) throws IgniteCheckedException {
+ if (!done.compareAndSet(false, true))
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591763373
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
##########
@@ -315,4 +347,183 @@ public void testClusterSnapshotCheckCRCFail() throws Exception {
Exception ex = res.exceptions().values().iterator().next();
assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
}
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedHashes() throws Exception {
+ int keys = 1;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] anton-vinogradov commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
anton-vinogradov commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r584524322
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
Review comment:
Wrong package for this class.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
Review comment:
Description missed at Javadoc.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(U.maskForFileName(consId)));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ File[] grps = nodePath.listFiles(path -> path.isDirectory() &&
Review comment:
Should this lookup have explicitly one result?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
Review comment:
Looks like the comment is not related to the 'store' field.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java
##########
@@ -371,7 +371,7 @@ private void setDirectCount(long pageAddr, int cnt) {
* @param pageAddr Page address.
* @return Direct count.
*/
- private int getDirectCount(long pageAddr) {
+ public int getDirectCount(long pageAddr) {
Review comment:
It is not a good idea to make something public for tests only
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
+ if (firstScan && freeSpace == 0 && rowsCnt == 1) {
Review comment:
Looks like you may combine all `(firstScan && ... )` to a single section.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
Review comment:
Should rows always be empty when finished?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
Review comment:
Since you create this on a group read what writes should be synchronized?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1113,6 +1136,59 @@ private static String snapshotMetaFileName(String consId) {
return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
}
+ /**
+ * @param snpName Snapshot name.
+ * @param grpName Cache group name.
+ * @param partId Partition id.
+ * @return Iterator over partition.
+ */
+ public GridCloseableIterator<CacheDataRow> partitionRows(String snpName,
+ String consId,
+ String grpName,
+ int partId
+ ) throws IgniteCheckedException {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ throw new IgniteCheckedException("Snapshot directory doesn't exists: " + snpDir.getAbsolutePath());
+
+ File nodePath = new File(snpDir, databaseRelativePath(U.maskForFileName(consId)));
+
+ if (!nodePath.exists())
+ throw new IgniteCheckedException("Consistent id directory doesn't exists: " + nodePath.getAbsolutePath());
+
+ File[] grps = nodePath.listFiles(path -> path.isDirectory() &&
+ (path.getName().equalsIgnoreCase(cacheDirName(true, grpName)) ||
+ path.getName().equalsIgnoreCase(cacheDirName(false, grpName))));
Review comment:
Could we simplify this by using the explicit isSharedFroup flag got from Group Description?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
Review comment:
Could we simplify this to something like
`Since the is no special marker for initial pages, but the goal is to detect all initial pages, marking all pages were referenced from others to keep only initial pages unmarked.`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
+ if (firstScan && freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ skipPages.touch(PageIdUtils.pageIndex(PageIdUtils.pageId(link)));
+
+ continue;
+ }
+
+ skipPages.touch(pageIdx);
+
+ for (int itemId = 0; itemId < rowsCnt; itemId++) {
+ DataRow row = new DataRow();
+
+ row.partition(partId);
+
+ row.initFromPageBuffer(
+ sctx,
+ coctx,
+ new IgniteThrowableFunction<Long, ByteBuffer>() {
+ @Override public ByteBuffer apply(Long nextPageId) throws IgniteCheckedException {
+ fragmentBuff.clear();
+
+ boolean read = store.read(nextPageId, fragmentBuff, true);
+
+ assert read : nextPageId;
+
+ // Fragment of page has been read, might be skipped further.
+ skipPages.touch(PageIdUtils.pageIndex(nextPageId));
Review comment:
Looks like this page already at `skipPages`
Removing this we able to use regular BitSet instead of the atomic one?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+
+import java.util.concurrent.atomic.AtomicIntegerArray;
+
+/**
+ * Atomic bitset array.
+ */
+class AtomicBitSet {
Review comment:
Do we really need BitSet to be atomic here, how many threads use them when the snapshot is loading?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AtomicBitSet.java
##########
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+
+import java.util.concurrent.atomic.AtomicIntegerArray;
+
+/**
+ * Atomic bitset array.
+ */
+class AtomicBitSet {
+ /**
+ * Container of bits.
+ */
+ private final AtomicIntegerArray arr;
+
+ /**
+ * Size of array of bits.
+ */
+ private final int size;
+
+ /**
+ * @param size Size of array.
+ */
+ public AtomicBitSet(int size) {
+ this.size = size;
+
+ arr = new AtomicIntegerArray((size + 31) >>> 5);
+ }
+
+ /**
+ * @return Size of bitset in bits.
+ */
+ public int size() {
+ return size;
+ }
+
+ /**
+ * @param idx Index in bitset array.
+ * @return {@code true} if the bit is set.
+ */
+ public boolean check(long idx) {
+ if (idx >= size)
+ return false;
+
+ int bit = 1 << idx;
+ int bucket = (int)(idx >>> 5);
+
+ int cur = arr.get(bucket);
+
+ return (cur & bit) == bit;
+ }
+
+ /**
+ * @param off Bit position to change.
+ * @return {@code true} if bit has been set,
+ * {@code false} if bit changed by another thread or out of range.
+ */
+ public boolean touch(long off) {
Review comment:
touch result never checked at "new code"
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
##########
@@ -223,6 +228,66 @@ public static String formatUpdateCountersDiff(IgniteEx ig, List<Integer> diff) {
return diff;
}
+ /**
+ * @param updCntr Partition update counter prior check.
+ * @param grpId Group id.
+ * @param partId Partition id.
+ * @param grpName Group name.
+ * @param consId Local node consistent id.
+ * @param state Partition state to check.
+ * @param isPrimary {@code true} if partition is primary.
+ * @param partSize Partition size on disk.
+ * @param it Iterator though partition data rows.
+ * @throws IgniteCheckedException If fails.
+ * @return Map of calculated partition.
+ */
+ public static Map<PartitionKeyV2, PartitionHashRecordV2> calculatePartitionHash(
+ long updCntr,
+ int grpId,
+ int partId,
+ String grpName,
+ Object consId,
+ GridDhtPartitionState state,
+ boolean isPrimary,
+ long partSize,
+ GridIterator<CacheDataRow> it
+ ) throws IgniteCheckedException {
+ int partHash = 0;
+
+ PartitionKeyV2 partKey = new PartitionKeyV2(grpId, partId, grpName);
+
+ if (state == GridDhtPartitionState.MOVING || state == GridDhtPartitionState.LOST) {
+ PartitionHashRecordV2 movingHashRecord = new PartitionHashRecordV2(
+ partKey,
+ isPrimary,
+ consId,
+ partHash,
+ updCntr,
+ state == GridDhtPartitionState.MOVING ? PartitionHashRecordV2.MOVING_PARTITION_SIZE : 0,
+ state == GridDhtPartitionState.MOVING ? PartitionHashRecordV2.PartitionState.MOVING : PartitionHashRecordV2.PartitionState.LOST
+ );
+
+ return Collections.singletonMap(partKey, movingHashRecord);
+ }
+
+ if (state != GridDhtPartitionState.OWNING)
+ return emptyMap();
+
+ while (it.hasNextX()) {
+ CacheDataRow row = it.nextX();
+
+ partHash += row.key().hashCode();
+
+ // Object context may be null since the value bytes have been read directly from page.
Review comment:
`may be null` -> `is not required`
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
+ if (firstScan && freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ skipPages.touch(PageIdUtils.pageIndex(PageIdUtils.pageId(link)));
+
+ continue;
+ }
+
+ skipPages.touch(pageIdx);
+
+ for (int itemId = 0; itemId < rowsCnt; itemId++) {
+ DataRow row = new DataRow();
+
+ row.partition(partId);
+
+ row.initFromPageBuffer(
+ sctx,
+ coctx,
+ new IgniteThrowableFunction<Long, ByteBuffer>() {
+ @Override public ByteBuffer apply(Long nextPageId) throws IgniteCheckedException {
+ fragmentBuff.clear();
+
+ boolean read = store.read(nextPageId, fragmentBuff, true);
+
+ assert read : nextPageId;
+
+ // Fragment of page has been read, might be skipped further.
+ skipPages.touch(PageIdUtils.pageIndex(nextPageId));
+
+ return fragmentBuff;
+ }
+ },
+ locBuff,
+ io,
+ itemId,
+ false,
+ CacheDataRowAdapter.RowData.FULL,
+ false);
+
+ rows.add(row);
+ }
+ }
+
+ if (currIdx == 2 * pages) {
+ finished = true;
Review comment:
Should be volatile in case more than one thread can use this method.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/tree/io/AbstractDataPageIO.java
##########
@@ -436,7 +436,7 @@ private boolean checkCount(int cnt) {
* @param pageAddr Page address.
* @return Indirect count.
*/
- private int getIndirectCount(long pageAddr) {
+ public int getIndirectCount(long pageAddr) {
Review comment:
It is not a good idea to make something public for tests only
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ // For pages which contains only the incomplete fragment of a data row
+ // the rowsCnt will always be equal to 1. Skip such pages and read them
+ // on the second iteration.
+ if (firstScan && freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ skipPages.touch(PageIdUtils.pageIndex(PageIdUtils.pageId(link)));
+
+ continue;
+ }
+
+ skipPages.touch(pageIdx);
Review comment:
Should we have "assert !firstScan" here?
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1347,198 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /** */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which must be skipped at the second iteration. */
+ private final AtomicBitSet skipPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration reached its end. */
+ private boolean finished;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx = -1;
+
+ /**
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ // CacheGroupContext may be null. It means that this cache group is under eviction for now.
+ // Since the persisted cache groups and their partitions are guarded by external machinery we
+ // can avoid it here.
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.sync();
+ pages = store.pages();
+ skipPages = new AtomicBitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + skipPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (finished && rows.isEmpty())
+ return false;
+
+ try {
+ while (rows.isEmpty() && ++currIdx < 2 * pages) {
+ int pageIdx = currIdx % pages;
+ boolean firstScan = currIdx < pages;
+
+ if (skipPages.check(pageIdx))
+ continue;
+
+ locBuff.clear();
+
+ long pageId = PageIdUtils.pageId(partId, PageIdAllocator.FLAG_DATA, pageIdx);
+ boolean success = store.read(pageId, locBuff, true);
+
+ assert success : PageIdUtils.toDetailString(pageId);
+
+ // Skip not data pages.
+ if (firstScan && PageIO.getType(locBuff) != T_DATA) {
+ skipPages.touch(pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = GridUnsafe.bufferAddress(locBuff);
+
+ DataPageIO io = PageIO.getPageIO(T_DATA, PageIO.getVersion(locBuff));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (firstScan && rowsCnt == 0) {
Review comment:
// Skip empty page.
comment will be useful
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591811149
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
##########
@@ -223,6 +228,66 @@ public static String formatUpdateCountersDiff(IgniteEx ig, List<Integer> diff) {
return diff;
}
+ /**
+ * @param updCntr Partition update counter prior check.
+ * @param grpId Group id.
+ * @param partId Partition id.
+ * @param grpName Group name.
+ * @param consId Local node consistent id.
+ * @param state Partition state to check.
+ * @param isPrimary {@code true} if partition is primary.
+ * @param partSize Partition size on disk.
+ * @param it Iterator though partition data rows.
+ * @throws IgniteCheckedException If fails.
+ * @return Map of calculated partition.
+ */
+ public static Map<PartitionKeyV2, PartitionHashRecordV2> calculatePartitionHash(
Review comment:
Agree, looks odd. I've tried to keep all these changes `as is` to simplify the review, however, it looks like the method signature can by simplified without huge refactoring.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591775767
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
Review comment:
Fixed.
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591748020
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
+ ByteBuffer fragmentBuff = reader.apply(pageId);
+
+ long fragmentAddr = GridUnsafe.bufferAddress(fragmentBuff);
+ DataPageIO io2 = PageIO.getPageIO(T_DATA, PageIO.getVersion(fragmentBuff));
+
+ incomplete = readIncomplete(incomplete, sctx, coctx, fragmentBuff.capacity(), fragmentBuff.capacity(),
+ fragmentAddr, itemId(nextLink), io2, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ nextLink = incomplete.getNextLink();
+ }
+ catch (Exception e) {
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CacheDataRowAdapter.java
##########
@@ -160,6 +166,61 @@ public final void initFromLink(
doInitFromLink(link, sharedCtx, coctx, pageMem, grpId, statHolder, readCacheId, rowData, null, skipVer);
}
+ /**
+ * @param io Data page IO.
+ * @param itemId Row item Id.
+ * @throws IgniteCheckedException If failed.
+ */
+ public final void initFromPageBuffer(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ IgniteThrowableFunction<Long, ByteBuffer> reader,
+ ByteBuffer pageBuff,
+ DataPageIO io,
+ int itemId,
+ boolean readCacheId,
+ RowData rowData,
+ boolean skipVer
+ ) throws IgniteCheckedException {
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ IncompleteObject<?> incomplete = readIncomplete(null, sctx, coctx, pageBuff.capacity(), pageBuff.capacity(),
+ pageAddr, itemId, io, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ long nextLink = incomplete.getNextLink();
+
+ if (nextLink == 0)
+ return;
+
+ do {
+ long pageId = pageId(nextLink);
+
+ try {
+ ByteBuffer fragmentBuff = reader.apply(pageId);
+
+ long fragmentAddr = GridUnsafe.bufferAddress(fragmentBuff);
+ DataPageIO io2 = PageIO.getPageIO(T_DATA, PageIO.getVersion(fragmentBuff));
+
+ incomplete = readIncomplete(incomplete, sctx, coctx, fragmentBuff.capacity(), fragmentBuff.capacity(),
+ fragmentAddr, itemId(nextLink), io2, rowData, readCacheId, skipVer);
+
+ if (incomplete == null)
+ return;
+
+ nextLink = incomplete.getNextLink();
+ }
+ catch (Exception e) {
+ throw new IgniteException("Error during reading DataRow [pageId=" + pageId + ']', e);
+ }
+ }
+ while (nextLink != 0);
+
+ assert isReady() : "ready";
Review comment:
Fixed.
##########
File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
##########
@@ -1271,6 +1352,230 @@ static void copy(FileIOFactory factory, File from, File to, long length) {
}
}
+ /**
+ * Ves pokrit assertami absolutely ves,
+ * PageScan iterator in the ignite core est.
+ */
+ private static class PageScanIterator extends GridCloseableIteratorAdapter<CacheDataRow> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Page store to iterate over. */
+ @GridToStringExclude
+ private final PageStore store;
+
+ /** Page store partition id. */
+ private final int partId;
+
+ /** Grid cache shared context. */
+ private final GridCacheSharedContext<?, ?> sctx;
+
+ /** Cache object context for key/value deserialization. */
+ private final CacheObjectContext coctx;
+
+ /** Buffer to read pages. */
+ private final ByteBuffer locBuff;
+
+ /** Buffer to read the rest part of fragmented rows. */
+ private final ByteBuffer fragmentBuff;
+
+ /** Total pages in the page store. */
+ private final int pages;
+
+ /** Pages which already marked and postponed to be read on the second iteration. */
+ private final BitSet markedPages;
+
+ /** Pages which already read and must be skipped. */
+ private final BitSet readPages;
+
+ /** Batch of rows read through iteration. */
+ private final Deque<CacheDataRow> rows = new LinkedList<>();
+
+ /** {@code true} if the iteration though partition reached its end. */
+ private boolean secondScanComplete;
+
+ /**
+ * Current partition page index for read. Due to we read the partition twice it
+ * can't be greater that 2 * store.size().
+ */
+ private int currIdx;
+
+ /**
+ * During scanning a cache partition presented as {@code PageStore} we must guarantee the following:
+ * all the pages of this storage remains unchanged during the Iterator remains opened, the stored data
+ * keeps its consistency. We can't read the {@code PageStore} during an ongoing checkpoint over it.
+ *
+ * @param coctx Cache object context.
+ * @param store Page store to read.
+ * @param partId Partition id.
+ * @throws IgniteCheckedException If fails.
+ */
+ public PageScanIterator(
+ GridCacheSharedContext<?, ?> sctx,
+ CacheObjectContext coctx,
+ PageStore store,
+ int partId
+ ) throws IgniteCheckedException {
+ this.store = store;
+ this.partId = partId;
+ this.coctx = coctx;
+ this.sctx = sctx;
+
+ store.ensure();
+ pages = store.pages();
+ markedPages = new BitSet(pages);
+ readPages = new BitSet(pages);
+
+ locBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ fragmentBuff = ByteBuffer.allocateDirect(store.getPageSize())
+ .order(ByteOrder.nativeOrder());
+ }
+
+ /** {@inheritDoc */
+ @Override protected CacheDataRow onNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ throw new NoSuchElementException("[partId=" + partId + ", store=" + store + ", skipPages=" + readPages + ']');
+
+ return rows.poll();
+ }
+
+ /** {@inheritDoc */
+ @Override protected boolean onHasNext() throws IgniteCheckedException {
+ if (secondScanComplete && rows.isEmpty())
+ return false;
+
+ try {
+ for (; currIdx < 2 * pages && rows.isEmpty(); currIdx++) {
+ boolean first = currIdx < pages;
+ int pageIdx = currIdx % pages;
+
+ if (readPages.get(pageIdx) || (!first && markedPages.get(pageIdx)))
+ continue;
+
+ if (!readPageFromStore(pageId(partId, FLAG_DATA, pageIdx), locBuff)) {
+ // Skip not FLAG_DATA pages.
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ long pageAddr = bufferAddress(locBuff);
+ DataPageIO io = getPageIO(T_DATA, getVersion(pageAddr));
+ int freeSpace = io.getFreeSpace(pageAddr);
+ int rowsCnt = io.getDirectCount(pageAddr);
+
+ if (first) {
+ // Skip empty pages.
+ if (rowsCnt == 0) {
+ changeBit(readPages, pageIdx);
+
+ continue;
+ }
+
+ // There is no difference between a page containing an incomplete DataRow fragment and
+ // the page where DataRow takes up all the free space. There is no a dedicated
+ // flag for this case in page header.
+ // During the storage scan we can skip such pages at the first iteration over the partition file,
+ // since all the fragmented pages will be marked by BitSet array we will safely read the others
+ // on the second iteration.
+ if (freeSpace == 0 && rowsCnt == 1) {
+ DataPagePayload payload = io.readPayload(pageAddr, 0, locBuff.capacity());
+
+ long link = payload.nextLink();
+
+ if (link != 0)
+ changeBit(markedPages, pageIndex(pageId(link)));
+
+ continue;
+ }
+ }
+
+ changeBit(readPages, pageIdx);
+
+ for (int itemId = 0; itemId < rowsCnt; itemId++) {
+ DataRow row = new DataRow();
+
+ row.partition(partId);
+
+ row.initFromPageBuffer(
+ sctx,
+ coctx,
+ new IgniteThrowableFunction<Long, ByteBuffer>() {
+ @Override public ByteBuffer apply(Long nextPageId) throws IgniteCheckedException {
+ boolean success = readPageFromStore(nextPageId, fragmentBuff);
+
+ assert success : "Only FLAG_DATA pages allowed: " + toDetailString(nextPageId);
+
+ // Fragment of page has been read, might be skipped further.
+ changeBit(readPages, pageIndex(nextPageId));
+
+ return fragmentBuff;
+ }
+ },
+ locBuff,
+ io,
+ itemId,
+ false,
+ CacheDataRowAdapter.RowData.FULL,
+ false);
+
+ rows.add(row);
+ }
+ }
+
+ if (currIdx == 2 * pages) {
+ secondScanComplete = true;
+
+ boolean set = true;
+
+ for (int j = 0; j < pages; j++)
+ set &= readPages.get(j);
+
+ assert set : "readPages=" + readPages + ", pages=" + pages;
+ }
+
+ return !rows.isEmpty();
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteCheckedException("Error during iteration through page store: " + this, e);
+ }
+ }
+
+ /**
+ * @param bitSet BitSet to change bit index.
+ * @param idx Index of bit to change.
+ */
+ private static void changeBit(BitSet bitSet, int idx) {
Review comment:
Fixed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [ignite] Mmuzaf commented on a change in pull request #8767: IGNITE-14134 commit snapshot partition iterator
Posted by GitBox <gi...@apache.org>.
Mmuzaf commented on a change in pull request #8767:
URL: https://github.com/apache/ignite/pull/8767#discussion_r591779861
##########
File path: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManagerSelfTest.java
##########
@@ -388,31 +403,266 @@ public void testLocalSnapshotOnCacheStopped() throws Exception {
snpFut.get(5_000, TimeUnit.MILLISECONDS);
}
- /**
- * @param src Source node to calculate.
- * @param grps Groups to collect owning parts.
- * @param rmtNodeId Remote node id.
- * @return Map of collected parts.
- */
- private static Map<Integer, Set<Integer>> owningParts(IgniteEx src, Set<Integer> grps, UUID rmtNodeId) {
- Map<Integer, Set<Integer>> result = new HashMap<>();
-
- for (Integer grpId : grps) {
- Set<Integer> parts = src.context()
- .cache()
- .cacheGroup(grpId)
- .topology()
- .partitions(rmtNodeId)
- .entrySet()
- .stream()
- .filter(p -> p.getValue() == GridDhtPartitionState.OWNING)
- .map(Map.Entry::getKey)
- .collect(Collectors.toSet());
-
- result.put(grpId, parts);
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorRandomizedLoader() throws Exception {
+ Random rnd = new Random();
+ int maxKey = 15_000;
+ int maxValSize = 32_768;
+ int loadingTimeMs = 60_000;
+
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>("tx1"))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithCache(1, CACHE_KEYS_RANGE, k -> new Value(new byte[1024]), ccfg);
+
+ IgniteCache<Integer, Value> cache = ignite.cache(ccfg.getName());
+
+ long startTime = U.currentTimeMillis();
+
+ IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (!Thread.currentThread().isInterrupted() && startTime + loadingTimeMs > U.currentTimeMillis()) {
+ if (rnd.nextBoolean())
+ cache.put(rnd.nextInt(15_000), new Value(new byte[rnd.nextInt(maxValSize)]));
+ else
+ cache.remove(rnd.nextInt(maxKey));
+ }
+
+ }, 5, "change-loader-");
+
+ fut.get();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Map<Integer, Value> iterated = new HashMap<>();
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ ccfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(ccfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ iterated.put(row.key().value(coctx, true), row.value().value(coctx, true));
+ }
+ }
+
+ stopAllGrids();
+
+ IgniteEx snpIgnite = startGridsFromSnapshot(1, SNAPSHOT_NAME);
+
+ IgniteCache<Integer, Value> snpCache = snpIgnite.cache(ccfg.getName());
+
+ assertEquals(snpCache.size(CachePeekMode.PRIMARY), iterated.size());
+ snpCache.forEach(e -> {
+ Value val = iterated.remove(e.getKey());
+
+ assertNotNull(val);
+ assertEquals(val.arr().length, e.getValue().arr().length);
+ });
+
+ assertTrue(iterated.isEmpty());
+ }
+
+ /** @throws Exception If fails */
+ @Test
+ public void testSnapshotIterator() throws Exception {
+ int keys = 127;
+
+ IgniteEx ignite = startGridsWithCache(2,
+ dfltCacheCfg.setAffinity(new RendezvousAffinityFunction(false, 1)), keys);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ // Invariant for cache: cache key always equals to cache value.
+ assertEquals("Invalid key/value pair [key=" + row.key() + ", val=" + row.value() + ']',
+ row.key().value(coctx, false, U.resolveClassLoader(ignite.configuration())),
+ (Integer)row.value().value(coctx, false));
+
+ rows++;
+ }
+ }
+
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorLargeRows() throws Exception {
+ int keys = 2;
+ CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
+ .setAffinity(new RendezvousAffinityFunction(false, 1));
+
+ IgniteEx ignite = startGridsWithoutCache(2);
+
+ for (int i = 0; i < keys; i++)
+ ignite.getOrCreateCache(ccfg).put(i, new Value(new byte[12008]));
+
+ forceCheckpoint();
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ int rows = 0;
+
+ try (GridCloseableIterator<CacheDataRow> iter = snp(ignite).partitionRows(SNAPSHOT_NAME,
+ ignite.context().pdsFolderResolver().resolveFolders().folderName(),
+ dfltCacheCfg.getName(),
+ 0)
+ ) {
+ CacheObjectContext coctx = ignite.cachex(dfltCacheCfg.getName()).context().cacheObjectContext();
+
+ while (iter.hasNext()) {
+ CacheDataRow row = iter.next();
+
+ assertEquals(12008, ((Value)row.value().value(coctx, false)).arr().length);
+ assertTrue((Integer)row.key().value(coctx, false, null) < 2);
+
+ rows++;
+ }
}
- return result;
+ assertEquals("Invalid number of rows: " + rows, keys, rows);
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testSnapshotIteratorDirectIndirectCountersWithCacheGroup() throws Exception {
Review comment:
Removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org