IGNITE-26295 Integrate Raft Log GC into the new storage (#7880)

Author: sashapolo

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/BeforeIndexFileCreatedCallback.java

@@ -0,0 +1,24 @@
+/*
+ * 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.raft.storage.segstore;
+
+/** Callback executed by the {@link RaftLogCheckpointer} before an index file is created. */
+@FunctionalInterface
+interface BeforeIndexFileCreatedCallback {
+    void beforeIndexFileCreated(long indexFileSize);
+}

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/FileProperties.java

@@ -22,7 +22,7 @@
 /**
  * Represents properties common for some file types (namely Segment and Index files) used by the log storage.
  */
-class FileProperties {
+class FileProperties implements Comparable<FileProperties> {
     /** File ordinal. Incremented each time a new file is created. */
     private final int ordinal;
 
@@ -54,6 +54,17 @@ int generation() {
         return generation;
     }
 
+    @Override
+    public int compareTo(FileProperties o) {
+        int cmp = Integer.compare(ordinal, o.ordinal);
+
+        if (cmp != 0) {
+            return cmp;
+        }
+
+        return Integer.compare(generation, o.generation);
+    }
+
     @Override
     public boolean equals(Object o) {
         if (o == null || getClass() != o.getClass()) {

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/IndexFileManager.java

@@ -51,6 +51,7 @@
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.VisibleForTesting;
 
 /**
  * File manager responsible for persisting {@link ReadModeIndexMemTable}s to index files.
@@ -215,10 +216,8 @@ private List<IndexMetaSpec> saveIndexMemtable(
         try (var os = new BufferedOutputStream(Files.newOutputStream(tmpFilePath, CREATE_NEW, WRITE))) {
             os.write(fileHeaderWithIndexMetas.header());
 
-            Iterator<Entry<Long, SegmentInfo>> it = indexMemTable.iterator();
-
-            while (it.hasNext()) {
-                SegmentInfo segmentInfo = it.next().getValue();
+            for (Entry<Long, SegmentInfo> longSegmentInfoEntry : indexMemTable) {
+                SegmentInfo segmentInfo = longSegmentInfoEntry.getValue();
 
                 // Segment Info may not contain payload in case of suffix truncation, see "IndexMemTable#truncateSuffix".
                 if (segmentInfo.size() > 0) {
@@ -385,11 +384,7 @@ private static FileHeaderWithIndexMetas serializeHeaderAndFillMetadata(
 
         var metaSpecs = new ArrayList<IndexMetaSpec>(numGroups);
 
-        Iterator<Entry<Long, SegmentInfo>> it = indexMemTable.iterator();
-
-        while (it.hasNext()) {
-            Entry<Long, SegmentInfo> entry = it.next();
-
+        for (Entry<Long, SegmentInfo> entry : indexMemTable) {
             // Using the boxed value to avoid unnecessary autoboxing later.
             Long groupId = entry.getKey();
 
@@ -504,6 +499,19 @@ private static byte[] payload(SegmentInfo segmentInfo) {
         return payloadBuffer.array();
     }
 
+    /**
+     * Computes the size in bytes that the index file for the given {@code indexMemTable} will occupy on disk.
+     */
+    static long computeIndexFileSize(ReadModeIndexMemTable indexMemTable) {
+        long total = headerSize(indexMemTable.numGroups());
+
+        for (Entry<Long, SegmentInfo> longSegmentInfoEntry : indexMemTable) {
+            total += payloadSize(longSegmentInfoEntry.getValue());
+        }
+
+        return total;
+    }
+
     private static int headerSize(int numGroups) {
         return COMMON_META_SIZE + numGroups * GROUP_META_SIZE;
     }
@@ -512,7 +520,8 @@ private static int payloadSize(SegmentInfo segmentInfo) {
         return segmentInfo.size() * Integer.BYTES;
     }
 
-    private static String indexFileName(FileProperties fileProperties) {
+    @VisibleForTesting
+    static String indexFileName(FileProperties fileProperties) {
         return String.format(INDEX_FILE_NAME_FORMAT, fileProperties.ordinal(), fileProperties.generation());
     }
 

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/MostGarbageFirstCompactionStrategy.java

@@ -0,0 +1,72 @@
+/*
+ * 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.raft.storage.segstore;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that prioritizes files with the most dead entries, fully-deletable files first.
+ */
+class MostGarbageFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+    private final Path segmentFilesDir;
+
+    private final IndexFileManager indexFileManager;
+
+    MostGarbageFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+        this.segmentFilesDir = segmentFilesDir;
+        this.indexFileManager = indexFileManager;
+    }
+
+    @Override
+    public Stream<FileProperties> selectCandidates() throws IOException {
+        var scores = new Object2LongOpenHashMap<FileProperties>();
+
+        Comparator<FileProperties> comparator =
+                Comparator.<FileProperties>comparingLong(props -> scores.computeIfAbsent(props, this::score))
+                        .thenComparing(Comparator.naturalOrder());
+
+        //noinspection resource
+        return Files.list(segmentFilesDir)
+                .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+                .map(SegmentFile::fileProperties)
+                .filter(props -> Files.exists(indexFileManager.indexFilePath(props)))
+                .sorted(comparator);
+    }
+
+    private long score(FileProperties props) {
+        Long2ObjectMap<IndexFileMeta> description = indexFileManager.describeSegmentFile(props.ordinal());
+
+        if (description.isEmpty()) {
+            return -1; // Fully deletable — highest priority.
+        }
+
+        long liveCount = 0;
+
+        for (IndexFileMeta meta : description.values()) {
+            liveCount += meta.lastLogIndexExclusive() - meta.firstLogIndexInclusive();
+        }
+
+        return liveCount;
+    }
+}

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/OldestFirstCompactionStrategy.java

@@ -0,0 +1,46 @@
+/*
+ * 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.raft.storage.segstore;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that yields fully-checkpointed segment files in write order (oldest first).
+ */
+class OldestFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+    private final Path segmentFilesDir;
+
+    private final IndexFileManager indexFileManager;
+
+    OldestFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+        this.segmentFilesDir = segmentFilesDir;
+        this.indexFileManager = indexFileManager;
+    }
+
+    @Override
+    public Stream<FileProperties> selectCandidates() throws IOException {
+        return Files.list(segmentFilesDir)
+                .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+                .map(SegmentFile::fileProperties)
+                .sorted()
+                .filter(props -> Files.exists(indexFileManager.indexFilePath(props)));
+    }
+}

modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogCheckpointer.java

@@ -21,7 +21,7 @@
 import static org.apache.ignite.internal.raft.storage.segstore.SegmentFileManager.SWITCH_SEGMENT_RECORD;
 import static org.apache.ignite.internal.raft.storage.segstore.SegmentInfo.MISSING_SEGMENT_FILE_OFFSET;
 import static org.apache.ignite.internal.util.IgniteUtils.closeAllManually;
-import static org.apache.ignite.lang.ErrorGroups.Marshalling.COMMON_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -62,14 +62,18 @@ class RaftLogCheckpointer {
 
     private final FailureProcessor failureProcessor;
 
+    private final BeforeIndexFileCreatedCallback beforeIndexFileCreated;
+
     RaftLogCheckpointer(
             String nodeName,
             IndexFileManager indexFileManager,
             FailureProcessor failureProcessor,
-            int maxQueueSize
+            int maxQueueSize,
+            BeforeIndexFileCreatedCallback beforeIndexFileCreated
     ) {
         this.indexFileManager = indexFileManager;
         this.failureProcessor = failureProcessor;
+        this.beforeIndexFileCreated = beforeIndexFileCreated;
 
         queue = new CheckpointQueue(maxQueueSize);
         checkpointThread = new IgniteThread(nodeName, "segstore-checkpoint", new CheckpointTask());
@@ -91,7 +95,7 @@ private void stopCheckpointThread() {
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
 
-            throw new IgniteInternalException(COMMON_ERR, "Interrupted while waiting for the checkpoint thread to finish.", e);
+            throw new IgniteInternalException(INTERNAL_ERR, "Interrupted while waiting for the checkpoint thread to finish.", e);
         }
     }
 
@@ -101,7 +105,7 @@ void onRollover(SegmentFile segmentFile, ReadModeIndexMemTable indexMemTable) {
         } catch (InterruptedException e) {
             Thread.currentThread().interrupt();
 
-            throw new IgniteInternalException(COMMON_ERR, "Interrupted while adding an entry to the checkpoint queue.", e);
+            throw new IgniteInternalException(INTERNAL_ERR, "Interrupted while adding an entry to the checkpoint queue.", e);
         }
     }
 
@@ -155,6 +159,11 @@ public void run() {
 
                     segmentFile.sync();
 
+                    long indexFileSize = IndexFileManager.computeIndexFileSize(entry.memTable());
+
+                    // Notify about the upcoming log size increase.
+                    beforeIndexFileCreated.beforeIndexFileCreated(indexFileSize);
+
                     indexFileManager.saveNewIndexMemtable(entry.memTable());
 
                     queue.removeHead();