diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java index 43db339d103fe..636c1e510f01f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java @@ -74,6 +74,7 @@ import org.apache.ignite.internal.processors.cache.persistence.CacheStripedExecutor; import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager; import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointProgress; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore; import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager; import org.apache.ignite.internal.processors.cache.persistence.file.FileVersionCheckingFactory; @@ -1759,30 +1760,14 @@ private void copyLocalAsync( ", snpDir=" + snpFile.getAbsolutePath() + ", name=" + snpFile.getName() + ']'); } - IgniteSnapshotManager.copy(snapMgr.ioFactory(), snpFile, tmpPartFile, snpFile.length()); + if (opCtx.isGroupCompressed(grpId)) + copyCompressed(opCtx, partFut.partId, snpFile, tmpPartFile); + else + IgniteSnapshotManager.copy(snapMgr.ioFactory(), snpFile, tmpPartFile, snpFile.length()); return tmpPartFile.toPath(); }, snapMgr.snapshotExecutorService()); - if (opCtx.isGroupCompressed(grpId)) { - copyPartFut = copyPartFut.thenComposeAsync( - p -> { - IgniteCompletableFuture result = new IgniteCompletableFuture<>(); - - try { - punchHole(grpId, partFut.partId, tmpPartFile); - - result.complete(tmpPartFile.toPath()); - } - catch (Throwable t) { - result.completeExceptionally(t); - } - return result; - }, - snapMgr.snapshotExecutorService() - ); - } - copyPartFut.whenComplete((r, t) -> opCtx.errHnd.accept(t)) .whenComplete((r, t) -> { if (t == null) @@ -1792,6 +1777,52 @@ private void copyLocalAsync( }); } + /** + * Copies a partition of a compressed cache group page by page writing only the compressed bytes of each page, so the + * rest of the page is left unallocated by the file system and the copy is sparse without punching holes in it. + */ + private void copyCompressed(SnapshotRestoreContext opCtx, int partId, File from, File to) { + FileIOFactory ioFactory = ctx.cache().context().snapshotMgr().ioFactory(); + int pageSize = ctx.cache().context().database().pageSize(); + + FileVersionCheckingFactory factory = new FileVersionCheckingFactory(ioFactory, ioFactory, () -> pageSize); + + try (FilePageStore src = (FilePageStore)factory.createPageStore(getTypeByPartId(partId), from, v -> {}); + FilePageStore dest = (FilePageStore)factory.createPageStore(getTypeByPartId(partId), to, v -> {})) { + src.init(); + dest.init(); + dest.beginRecover(); + + ByteBuffer buf = locBuff.get(); + + long pageId = PageIdUtils.pageId(partId, (byte)0, 0); + + for (int pageNo = 0; pageNo < src.pages(); pageId++, pageNo++) { + if (opCtx.stopChecker.getAsBoolean()) + throw new IgniteInterruptedException("The operation has been stopped on copy file: " + from.getAbsolutePath()); + + if (Thread.interrupted()) + throw new IgniteInterruptedException("Thread has been interrupted: " + Thread.currentThread().getName()); + + buf.clear(); + + src.read(pageId, buf, true); + + buf.rewind(); // The CRC check moves the position. + + if (PageIO.getCompressionType(buf) != CompressionProcessor.UNCOMPRESSED_PAGE) + buf.limit(PageIO.getCompressedSize(buf)); + + dest.write(pageId, buf, 0, false); + } + + dest.finishRecover(); + } + catch (IOException | IgniteCheckedException e) { + throw new IgniteException(e); + } + } + /** */ private void punchHole(int grpId, int partId, File partFile) throws Exception { FilePageStoreManager storeMgr = (FilePageStoreManager)ctx.cache().context().pageStore();