IGNITE-17177 Incremental Snaphots implementation

modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java

@@ -204,6 +204,8 @@ public static void main(String[] args) throws Exception {
 //        gen.generateAndWrite(GridDhtTxFinishRequest.class);
 //        gen.generateAndWrite(GridDhtTxFinishResponse.class);
 //
+//        gen.generateAndWrite(IncrementalSnapshotAwareMessage.class);
+
 //        gen.generateAndWrite(GridCacheTxRecoveryRequest.class);
 //        gen.generateAndWrite(GridCacheTxRecoveryResponse.class);
 

modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/indexreader/IgniteIndexReader.java

@@ -18,10 +18,7 @@
 package org.apache.ignite.internal.commandline.indexreader;
 
 import java.io.File;
-import java.io.IOException;
-import java.io.ObjectInputStream;
 import java.nio.ByteBuffer;
-import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -61,6 +58,7 @@
 import org.apache.ignite.internal.commandline.systemview.SystemViewCommand;
 import org.apache.ignite.internal.pagemem.PageIdAllocator;
 import org.apache.ignite.internal.pagemem.PageUtils;
+import org.apache.ignite.internal.processors.cache.GridLocalConfigManager;
 import org.apache.ignite.internal.processors.cache.StoredCacheData;
 import org.apache.ignite.internal.processors.cache.persistence.IndexStorageImpl;
 import org.apache.ignite.internal.processors.cache.persistence.StorageException;
@@ -116,7 +114,6 @@
 import static org.apache.ignite.internal.pagemem.PageIdUtils.pageId;
 import static org.apache.ignite.internal.pagemem.PageIdUtils.pageIndex;
 import static org.apache.ignite.internal.pagemem.PageIdUtils.partId;
-import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.INDEX_FILE_NAME;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.PART_FILE_TEMPLATE;
 import static org.apache.ignite.internal.util.GridUnsafe.allocateBuffer;
@@ -283,13 +280,13 @@ public IgniteIndexReader(
         for (int i = 0; i < partCnt; i++)
             partStores[i] = filePageStore(i, FLAG_DATA, storeFactory);
 
-        Arrays.stream(root.listFiles(f -> f.getName().endsWith(CACHE_DATA_FILENAME))).forEach(f -> {
-            try (ObjectInputStream stream = new ObjectInputStream(Files.newInputStream(f.toPath()))) {
-                StoredCacheData data = (StoredCacheData)stream.readObject();
+        Arrays.stream(FilePageStoreManager.cacheDataFiles(root)).forEach(f -> {
+            try {
+                StoredCacheData data = GridLocalConfigManager.readCacheData(f, null, null);
 
                 storedCacheData.put(CU.cacheId(data.config().getName()), data);
             }
-            catch (ClassNotFoundException | IOException e) {
+            catch (IgniteCheckedException e) {
                 log.warning("Can't read stored cache data. Inline for this cache will not be analyzed [f=" + f.getName() + ']', e);
             }
         });

modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCreateCommand.java

@@ -29,6 +29,7 @@
 import static org.apache.ignite.internal.commandline.CommandList.SNAPSHOT;
 import static org.apache.ignite.internal.commandline.CommandLogger.optional;
 import static org.apache.ignite.internal.commandline.snapshot.SnapshotCreateCommandOption.DESTINATION;
+import static org.apache.ignite.internal.commandline.snapshot.SnapshotCreateCommandOption.INCREMENTAL;
 import static org.apache.ignite.internal.commandline.snapshot.SnapshotCreateCommandOption.SYNC;
 
 /**
@@ -45,6 +46,7 @@ protected SnapshotCreateCommand() {
         String snpName = argIter.nextArg("Expected snapshot name.");
         String snpPath = null;
         boolean sync = false;
+        boolean incremental = false;
 
         while (argIter.hasNextSubArg()) {
             String arg = argIter.nextArg(null);
@@ -72,10 +74,15 @@ else if (option == SYNC) {
 
                 sync = true;
             }
+            else if (option == INCREMENTAL) {
+                if (incremental)
+                    throw new IllegalArgumentException(INCREMENTAL.argName() + " arg specified twice.");
 
+                incremental = true;
+            }
         }
 
-        cmdArg = new VisorSnapshotCreateTaskArg(snpName, snpPath, sync);
+        cmdArg = new VisorSnapshotCreateTaskArg(snpName, snpPath, sync, incremental);
     }
 
     /** {@inheritDoc} */
@@ -84,8 +91,9 @@ else if (option == SYNC) {
 
         params.put(DESTINATION.argName() + " " + DESTINATION.arg(), DESTINATION.description());
         params.put(SYNC.argName(), SYNC.description());
+        params.put(INCREMENTAL.argName(), INCREMENTAL.description());
 
         usage(log, "Create cluster snapshot:", SNAPSHOT, params, name(), SNAPSHOT_NAME_ARG,
-            optional(DESTINATION.argName(), DESTINATION.arg()), optional(SYNC.argName()));
+            optional(DESTINATION.argName(), DESTINATION.arg()), optional(SYNC.argName()), optional(INCREMENTAL.argName()));
     }
 }

modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCreateCommandOption.java

@@ -30,7 +30,11 @@ public enum SnapshotCreateCommandOption implements CommandArg {
 
     /** Snapshot directory path. */
     DESTINATION("--dest", "path", "Path to the directory where the snapshot will be saved. If not specified, " +
-        "the default configured snapshot directory will be used.");
+        "the default configured snapshot directory will be used."),
+
+    /** Incremental snapshot flag. */
+    INCREMENTAL("--incremental", null, "Create an incremental snapshot for previously created full snapshot. " +
+        "Full snapshot must be accessible via --dest and snapshot_name.");
 
     /** Name. */
     private final String name;

modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotSubcommand.java

@@ -87,7 +87,10 @@ protected SnapshotSubcommand(String name, Class<?> taskCls) {
      * @return General usage options.
      */
     protected Map<String, String> generalUsageOptions() {
-        return F.asMap(SNAPSHOT_NAME_ARG, "Snapshot name.");
+        return F.asMap(
+            SNAPSHOT_NAME_ARG,
+            "Snapshot name. In case incremental snapshot (--incremental) full snapshot name must be provided."
+        );
     }
 
     /**

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java

@@ -3163,7 +3163,7 @@ private void doClusterSnapshotCreate(boolean syncMode) throws Exception {
 
         // Invalid command syntax check.
         assertEquals(EXIT_CODE_INVALID_ARGUMENTS, execute(h, "--snapshot", "create", snpName, "blah"));
-        assertContains(log, testOut.toString(), "Invalid argument: blah. Possible options: --sync, --dest.");
+        assertContains(log, testOut.toString(), "Invalid argument: blah. Possible options: --sync, --dest, --incremental.");
 
         assertEquals(EXIT_CODE_INVALID_ARGUMENTS, execute(h, "--snapshot", "create", snpName, "--sync", "blah"));
         assertContains(log, testOut.toString(), "Invalid argument: blah.");

modules/core/src/main/java/org/apache/ignite/IgniteSnapshot.java

@@ -40,10 +40,18 @@ public interface IgniteSnapshot {
      * Create a consistent copy of all persistence cache groups from the whole cluster.
      *
      * @param name Snapshot unique name which satisfies the following name pattern [a-zA-Z0-9_].
-     * @return Future which will be completed when a process ends.
+     * @return Future which will be completed when the process ends.
      */
     public IgniteFuture<Void> createSnapshot(String name);
 
+    /**
+     * Create an incremental snapshot for a given full snapshot.
+     *
+     * @param fullSnapshot Snapshot unique name which satisfies the following name pattern [a-zA-Z0-9_].
+     * @return Future which will be completed when the process ends.
+     */
+    public IgniteFuture<Void> createIncrementalSnapshot(String fullSnapshot);
+
     /**
      * Cancel running snapshot operation. All intermediate results of cancelled snapshot operation will be deleted.
      * If snapshot already created this command will have no effect.
@@ -65,6 +73,19 @@ public interface IgniteSnapshot {
      */
     public IgniteFuture<Void> restoreSnapshot(String name, @Nullable Collection<String> cacheGroupNames);
 
+    /**
+     * Restore cache group(s) from the incremental snapshot.
+     * <p>
+     * <b>NOTE:</b> Cache groups to be restored from the snapshot must not present in the cluster, if they present,
+     * they must be destroyed by the user (eg with {@link IgniteCache#destroy()}) before starting this operation.
+     *
+     * @param name Snapshot name.
+     * @param cacheGroupNames Cache groups to be restored or {@code null} to restore all cache groups from the snapshot.
+     * @param incIdx Index of incremental snapshot.
+     * @return Future which will be completed when restore operation finished.
+     */
+    public IgniteFuture<Void> restoreIncrementalSnapshot(String name, @Nullable Collection<String> cacheGroupNames, int incIdx);
+
     /**
      * Cancel snapshot restore operation.
      *

modules/core/src/main/java/org/apache/ignite/internal/cdc/CdcMain.java

@@ -52,6 +52,7 @@
 import org.apache.ignite.internal.cdc.WalRecordsConsumer.DataEntryIterator;
 import org.apache.ignite.internal.processors.cache.GridLocalConfigManager;
 import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver;
 import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings;
 import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
@@ -79,7 +80,6 @@
 import static org.apache.ignite.internal.binary.BinaryUtils.METADATA_FILE_SUFFIX;
 import static org.apache.ignite.internal.pagemem.wal.record.WALRecord.RecordType.DATA_RECORD_V2;
 import static org.apache.ignite.internal.processors.cache.GridCacheUtils.UTILITY_CACHE_NAME;
-import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DIR_PREFIX;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_GRP_DIR_PREFIX;
 import static org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager.WAL_SEGMENT_FILE_FILTER;
@@ -705,7 +705,7 @@ private void updateCaches() {
                 .filter(File::exists)
                 // Cache group directory can contain several cache data files.
                 // See GridLocalConfigManager#cacheConfigurationFile(CacheConfiguration<?, ?>)
-                .flatMap(cacheDir -> Arrays.stream(cacheDir.listFiles(f -> f.getName().endsWith(CACHE_DATA_FILENAME))))
+                .flatMap(cacheDir -> Arrays.stream(FilePageStoreManager.cacheDataFiles(cacheDir)))
                 .map(f -> {
                     try {
                         CdcCacheEvent evt = GridLocalConfigManager.readCacheData(

modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java

@@ -136,6 +136,7 @@
 import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccRecoveryFinishedMessage;
 import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccSnapshotResponse;
 import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccTxSnapshotRequest;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessage;
 import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessage;
 import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
@@ -382,6 +383,9 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         factory.register(SnapshotFilesRequestMessage.TYPE_CODE, SnapshotFilesRequestMessage::new);
         factory.register(SnapshotFilesFailureMessage.TYPE_CODE, SnapshotFilesFailureMessage::new);
 
+        // Incremental snapshot.
+        factory.register(IncrementalSnapshotAwareMessage.TYPE_CODE, IncrementalSnapshotAwareMessage::new);
+
         // Index statistics.
         factory.register(StatisticsKeyMessage.TYPE_CODE, StatisticsKeyMessage::new);
         factory.register(StatisticsDecimalMessage.TYPE_CODE, StatisticsDecimalMessage::new);
@@ -390,8 +394,11 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
         factory.register(StatisticsRequest.TYPE_CODE, StatisticsRequest::new);
         factory.register(StatisticsResponse.TYPE_CODE, StatisticsResponse::new);
 
-        // [-3..119] [124..129] [-23..-28] [-36..-55] - this
+        // [-3..119] [124..129] [-23..-28] [-36..-55] [183..188] - this
         // [120..123] - DR
+        // [-44, 0..2, 42, 200..204, 210, 302] - Use in tests.
+        // [300..307, 350..352] - CalciteMessageFactory.
+        // [400] - Incremental snapshot.
         // [-4..-22, -30..-35, -54..-57] - SQL
         // [2048..2053] - Snapshots
         // [-42..-37] - former hadoop.

modules/core/src/main/java/org/apache/ignite/internal/managers/systemview/walker/SnapshotViewWalker.java

@@ -33,6 +33,7 @@ public class SnapshotViewWalker implements SystemViewRowAttributeWalker<Snapshot
         v.accept(1, "consistentId", String.class);
         v.accept(2, "baselineNodes", String.class);
         v.accept(3, "cacheGroups", String.class);
+        v.accept(4, "snapshotRecordSegment", Long.class);
     }
 
     /** {@inheritDoc} */
@@ -41,10 +42,11 @@ public class SnapshotViewWalker implements SystemViewRowAttributeWalker<Snapshot
         v.accept(1, "consistentId", String.class, row.consistentId());
         v.accept(2, "baselineNodes", String.class, row.baselineNodes());
         v.accept(3, "cacheGroups", String.class, row.cacheGroups());
+        v.accept(4, "snapshotRecordSegment", Long.class, row.snapshotRecordSegment());
     }
 
     /** {@inheritDoc} */
     @Override public int count() {
-        return 4;
+        return 5;
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/IgniteWriteAheadLogManager.java

@@ -17,9 +17,11 @@
 
 package org.apache.ignite.internal.pagemem.wal;
 
+import java.io.File;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.pagemem.wal.record.RolloverType;
 import org.apache.ignite.internal.pagemem.wal.record.WALRecord;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedManager;
@@ -231,4 +233,25 @@ public WALIterator replay(
      * Start automatically releasing segments when reaching {@link DataStorageConfiguration#getMaxWalArchiveSize()}.
      */
     void startAutoReleaseSegments();
+
+    /**
+     * Archive directory if any.
+     *
+     * @return Archive directory.
+     */
+    @Nullable File archiveDir();
+
+    /**
+     * @param idx Segment index.
+     * @return Compressed archive segment.
+     */
+    @Nullable File compactedSegment(long idx);
+
+    /**
+     * Blocks current thread while segment with the {@code idx} not compressed.
+     * If segment compressed, already, returns immediately.
+     *
+     * @param idx Segment index.
+     */
+    void awaitCompacted(long idx) throws IgniteInterruptedCheckedException;
 }

modules/core/src/main/java/org/apache/ignite/internal/pagemem/wal/record/IncrementalSnapshotFinishRecord.java

@@ -0,0 +1,101 @@
+/*
+ * 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.pagemem.wal.record;
+
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.CacheVersionIO;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * This record is written to WAL after incremental snapshot finished on a baseline node.
+ * <p>
+ * During recovery node must apply:
+ * 1. Transactions committed before {@link IncrementalSnapshotStartRecord} except those contained in {@link #excluded()}.
+ * 2. Transactions committed between {@link IncrementalSnapshotStartRecord} and {@link IncrementalSnapshotFinishRecord}
+ *    and contained in {@link #included()}.
+ */
+public class IncrementalSnapshotFinishRecord extends WALRecord {
+    /** Incremental snapshot ID. */
+    @GridToStringInclude
+    private final UUID id;
+
+    /**
+     * Set of transactions committed between {@link IncrementalSnapshotStartRecord} and {@link IncrementalSnapshotFinishRecord}
+     * to include into the incremental snapshot.
+     */
+    @GridToStringInclude
+    private final Set<GridCacheVersion> included;
+
+    /**
+     * Set of transactions committed before {@link IncrementalSnapshotStartRecord} to exclude from the incremental snapshot.
+     */
+    @GridToStringInclude
+    private final Set<GridCacheVersion> excluded;
+
+    /** */
+    public IncrementalSnapshotFinishRecord(UUID id, Set<GridCacheVersion> included, Set<GridCacheVersion> excluded) {
+        this.id = id;
+        this.included = included;
+        this.excluded = excluded;
+    }
+
+    /** */
+    public Set<GridCacheVersion> included() {
+        return included;
+    }
+
+    /** */
+    public Set<GridCacheVersion> excluded() {
+        return excluded;
+    }
+
+    /** */
+    public UUID id() {
+        return id;
+    }
+
+    /** {@inheritDoc} */
+    @Override public RecordType type() {
+        return RecordType.INCREMENTAL_SNAPSHOT_FINISH_RECORD;
+    }
+
+    /**
+     * Calculating the size of the record.
+     *
+     * @return Size in bytes.
+     */
+    public int dataSize() {
+        int size = 16 + 4 + 4;  // ID, included and excluded tx count.
+
+        for (GridCacheVersion v: included)
+            size += CacheVersionIO.size(v, false);
+
+        for (GridCacheVersion v: excluded)
+            size += CacheVersionIO.size(v, false);
+
+        return size;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(IncrementalSnapshotFinishRecord.class, this);
+    }
+}