TC debug snp compression

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<Path> 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();