You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2025/11/28 09:07:11 UTC
(ignite-3) branch ignite-26998 created (now ba4127bc963)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a change to branch ignite-26998
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at ba4127bc963 IGNITE-26998 wip
This branch includes the following new commits:
new cfe271baa46 IGNITE-26998 wip
new 1372be5be55 IGNITE-26998 wip
new a8a6c83d11e IGNITE-26998 wip
new ba4127bc963 IGNITE-26998 wip
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 02/04: IGNITE-26998 wip
Posted by tk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch ignite-26998
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit 1372be5be5542fadade4a66905c1c92aa3b3179d
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Thu Nov 27 18:13:53 2025 +0300
IGNITE-26998 wip
---
.../replicator/raft/snapshot/PartitionDataStorage.java | 7 ++++---
.../ignite/internal/storage/MvPartitionStorage.java | 9 +++++----
.../storage/ThreadAssertingMvPartitionStorage.java | 4 ++--
.../internal/storage/BaseMvPartitionStorageTest.java | 7 +++++--
.../internal/storage/impl/TestMvPartitionStorage.java | 9 +++++----
.../mv/AbstractPageMemoryMvPartitionStorage.java | 9 +++++----
.../ignite/internal/storage/rocksdb/GarbageCollector.java | 15 ++++++++-------
.../storage/rocksdb/RocksDbMvPartitionStorage.java | 3 ++-
.../internal/table/distributed/gc/GcUpdateHandler.java | 9 ++++++---
.../raft/snapshot/SnapshotAwarePartitionDataStorage.java | 4 ++--
.../table/distributed/gc/AbstractGcUpdateHandlerTest.java | 6 +++---
.../ignite/distributed/TestPartitionDataStorage.java | 5 +++--
12 files changed, 50 insertions(+), 37 deletions(-)
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
index 7ce61504bf1..4f183843a17 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionDataStorage.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.partition.replicator.raft.snapshot;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.close.ManuallyCloseable;
@@ -246,11 +247,11 @@ public interface PartitionDataStorage extends ManuallyCloseable {
PartitionTimestampCursor scan(HybridTimestamp timestamp) throws StorageException;
/**
- * Returns the head of GC queue.
+ * Returns entries from the queue starting from the head.
*
- * @see MvPartitionStorage#peek(HybridTimestamp)
+ * @see MvPartitionStorage#peek
*/
- @Nullable GcEntry peek(HybridTimestamp lowWatermark);
+ List<GcEntry> peek(HybridTimestamp lowWatermark, int count);
/**
* Delete GC entry from the GC queue and corresponding version chain.
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
index cdbc31fe921..da4cf7dc8ba 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/MvPartitionStorage.java
@@ -302,17 +302,18 @@ public interface MvPartitionStorage extends ManuallyCloseable {
List<RowMeta> rowsStartingWith(RowId lowerBoundInclusive, RowId upperBoundInclusive, int limit) throws StorageException;
/**
- * Returns the head of GC queue.
+ * Returns entries from the queue starting from the head.
*
* @param lowWatermark Upper bound for commit timestamp of GC entry, inclusive.
- * @return Queue head or {@code null} if there are no entries below passed low watermark.
+ * @param count Requested count of entries.
+ * @return First entries in the GC queue that are less than or equal to passed low watermark.
*/
- @Nullable GcEntry peek(HybridTimestamp lowWatermark);
+ List<GcEntry> peek(HybridTimestamp lowWatermark, int count);
/**
* Delete GC entry from the GC queue and corresponding version chain. Row ID of the entry must be locked to call this method.
*
- * @param entry Entry, previously returned by {@link #peek(HybridTimestamp)}.
+ * @param entry Entry, previously returned by {@link #peek}.
* @return Polled binary row, or {@code null} if the entry has already been deleted by another thread.
*
* @see Locker#lock(RowId)
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/ThreadAssertingMvPartitionStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/ThreadAssertingMvPartitionStorage.java
index 93b58b9b577..9d83df999d9 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/ThreadAssertingMvPartitionStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/ThreadAssertingMvPartitionStorage.java
@@ -166,10 +166,10 @@ public class ThreadAssertingMvPartitionStorage implements MvPartitionStorage, Wr
}
@Override
- public @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
+ public List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
assertThreadAllowsToRead();
- return partitionStorage.peek(lowWatermark);
+ return partitionStorage.peek(lowWatermark, count);
}
@Override
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvPartitionStorageTest.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvPartitionStorageTest.java
index c5d6ba98e63..0959e136c6d 100644
--- a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvPartitionStorageTest.java
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/BaseMvPartitionStorageTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.storage;
+import java.util.List;
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.BinaryRow;
@@ -159,12 +160,14 @@ public abstract class BaseMvPartitionStorageTest extends BaseMvStoragesTest {
@Nullable BinaryRowAndRowId pollForVacuum(HybridTimestamp lowWatermark) {
while (true) {
BinaryRowAndRowId binaryRowAndRowId = storage.runConsistently(locker -> {
- GcEntry gcEntry = storage.peek(lowWatermark);
+ List<GcEntry> gcEntries = storage.peek(lowWatermark, 1);
- if (gcEntry == null) {
+ if (gcEntries.isEmpty()) {
return null;
}
+ GcEntry gcEntry = gcEntries.get(0);
+
locker.lock(gcEntry.getRowId());
return new BinaryRowAndRowId(storage.vacuum(gcEntry), gcEntry.getRowId());
diff --git a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvPartitionStorage.java b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvPartitionStorage.java
index 4666c5e59e3..0f6b8a7d3a2 100644
--- a/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvPartitionStorage.java
+++ b/modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/impl/TestMvPartitionStorage.java
@@ -698,17 +698,18 @@ public class TestMvPartitionStorage implements MvPartitionStorage {
}
@Override
- public synchronized @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
+ // TODO: IGNITE-26998 Переделать
+ public synchronized List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
assert THREAD_LOCAL_LOCKER.get() != null;
try {
VersionChain versionChain = gcQueue.first();
- if (versionChain.ts.compareTo(lowWatermark) > 0) {
- return null;
+ if (versionChain == null || versionChain.ts.compareTo(lowWatermark) > 0) {
+ return List.of();
}
- return versionChain;
+ return List.of(versionChain);
} catch (NoSuchElementException e) {
return null;
}
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
index c5b4c2a4e78..3964c85eedb 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
@@ -962,7 +962,8 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
}
@Override
- public @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
+ // TODO: IGNITE-26998 Исправить реализацию
+ public List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
assert THREAD_LOCAL_LOCKER.get() != null;
// Assertion above guarantees that we're in "runConsistently" closure.
@@ -972,17 +973,17 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
// Garbage collection queue is empty.
if (head == null) {
- return null;
+ return List.of();
}
HybridTimestamp rowTimestamp = head.getTimestamp();
// There are no versions in the garbage collection queue before watermark.
if (rowTimestamp.compareTo(lowWatermark) > 0) {
- return null;
+ return List.of();
}
- return head;
+ return List.of(head);
}
@Override
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
index d6eeac69064..8b2624d174c 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/GarbageCollector.java
@@ -35,8 +35,10 @@ import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.ROW
import static org.apache.ignite.internal.storage.rocksdb.RocksDbStorageUtils.TABLE_ID_SIZE;
import java.nio.ByteBuffer;
+import java.util.List;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.BinaryRow;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.gc.GcEntry;
import org.jetbrains.annotations.Nullable;
@@ -173,13 +175,12 @@ class GarbageCollector {
}
/**
- * Polls an element for vacuum. See {@link org.apache.ignite.internal.storage.MvPartitionStorage#peek(HybridTimestamp)}.
+ * Polls an elements for vacuum. See {@link MvPartitionStorage#peek}.
*
* @param writeBatch Current Write Batch.
* @param lowWatermark Low watermark.
- * @return Garbage collected element descriptor.
*/
- @Nullable GcEntry peek(WriteBatchWithIndex writeBatch, HybridTimestamp lowWatermark) {
+ List<GcEntry> peek(WriteBatchWithIndex writeBatch, HybridTimestamp lowWatermark) {
// We retrieve the first element of the GC queue and seek for it in the data CF.
// However, the element that we need to garbage collect is the next (older one) element.
// First we check if there's anything to garbage collect. If the element is a tombstone we remove it.
@@ -189,7 +190,7 @@ class GarbageCollector {
if (invalid(gcIt)) {
// GC queue is empty.
- return null;
+ return List.of();
}
ByteBuffer gcKeyBuffer = readGcKey(gcIt);
@@ -198,16 +199,16 @@ class GarbageCollector {
if (gcRowVersion.getTimestamp().compareTo(lowWatermark) > 0) {
// No elements to garbage collect.
- return null;
+ return List.of();
}
- return gcRowVersion;
+ return List.of(gcRowVersion);
}
}
/**
- * Polls an element for vacuum. See {@link org.apache.ignite.internal.storage.MvPartitionStorage#vacuum(GcEntry)}.
+ * Polls an element for vacuum. See {@link MvPartitionStorage#vacuum(GcEntry)}.
*
* @param batch Write batch.
* @param entry Entry, previously returned by {@link #peek}.
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
index 38676f31b88..2ca56d14e98 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMvPartitionStorage.java
@@ -1290,7 +1290,8 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
}
@Override
- public @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
+ // TODO: IGNITE-26988 Переделать
+ public List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
WriteBatchWithIndex batch = requireWriteBatch();
// No busy lock required, we're already in "runConsistently" closure.
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
index 5b239514e23..c5ab546b188 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.table.distributed.gc;
+import java.util.List;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionDataStorage;
import org.apache.ignite.internal.schema.BinaryRow;
@@ -75,7 +76,7 @@ public class GcUpdateHandler {
return true;
}
- IntHolder countHolder = new IntHolder(count);
+ var countHolder = new IntHolder(count);
while (countHolder.get() > 0) {
VacuumResult vacuumResult = internalVacuumBatch(lowWatermark, countHolder);
@@ -131,12 +132,14 @@ public class GcUpdateHandler {
return VacuumResult.SHOULD_RELEASE;
}
- GcEntry gcEntry = storage.peek(lowWatermark);
+ // TODO: IGNITE-26998 Переделать а пока заглушка
+ List<GcEntry> gcEntries = storage.peek(lowWatermark, 1);
- if (gcEntry == null) {
+ if (gcEntries.isEmpty()) {
return VacuumResult.NO_GARBAGE_LEFT;
}
+ GcEntry gcEntry = gcEntries.get(0);
RowId rowId = gcEntry.getRowId();
if (useTryLock) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
index 6eaeea5b58d..507f0ea49e2 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/SnapshotAwarePartitionDataStorage.java
@@ -244,8 +244,8 @@ public class SnapshotAwarePartitionDataStorage implements PartitionDataStorage {
}
@Override
- public @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
- return partitionStorage.peek(lowWatermark);
+ public List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
+ return partitionStorage.peek(lowWatermark, count);
}
@Override
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
index 0f8fcd8db21..61b130b62a1 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
@@ -90,7 +90,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
HybridTimestamp lowWatermark = HybridTimestamp.MAX_VALUE;
assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 1));
- verify(partitionStorage).peek(lowWatermark);
+ verify(partitionStorage).peek(eq(lowWatermark), eq(1));
// Let's check that StorageUpdateHandler#vacuumBatch returns true.
clearInvocations(partitionStorage);
@@ -102,7 +102,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
addWriteCommitted(partitionStorage, rowId, row, clock.now());
assertTrue(gcUpdateHandler.vacuumBatch(lowWatermark, 1));
- verify(partitionStorage).peek(lowWatermark);
+ verify(partitionStorage).peek(eq(lowWatermark), eq(1));
verify(indexUpdateHandler).tryRemoveFromIndexes(any(), eq(rowId), any(), isNull());
}
@@ -129,7 +129,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 5));
- verify(partitionStorage, times(3)).peek(lowWatermark);
+ verify(partitionStorage, times(3)).peek(eq(lowWatermark), eq(5));
verify(indexUpdateHandler).tryRemoveFromIndexes(any(), eq(rowId0), any(), isNull());
verify(indexUpdateHandler).tryRemoveFromIndexes(any(), eq(rowId1), any(), isNull());
}
diff --git a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/TestPartitionDataStorage.java b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/TestPartitionDataStorage.java
index d446e5445db..4d68eb2a087 100644
--- a/modules/table/src/testFixtures/java/org/apache/ignite/distributed/TestPartitionDataStorage.java
+++ b/modules/table/src/testFixtures/java/org/apache/ignite/distributed/TestPartitionDataStorage.java
@@ -17,6 +17,7 @@
package org.apache.ignite.distributed;
+import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.hlc.HybridTimestamp;
@@ -167,8 +168,8 @@ public class TestPartitionDataStorage implements PartitionDataStorage {
}
@Override
- public @Nullable GcEntry peek(HybridTimestamp lowWatermark) {
- return partitionStorage.peek(lowWatermark);
+ public List<GcEntry> peek(HybridTimestamp lowWatermark, int count) {
+ return partitionStorage.peek(lowWatermark, count);
}
@Override
(ignite-3) 01/04: IGNITE-26998 wip
Posted by tk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch ignite-26998
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit cfe271baa46adf7ef23b21b4b0762dc6f8dbf576
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Thu Nov 27 15:03:39 2025 +0300
IGNITE-26998 wip
---
.../table/distributed/gc/GcUpdateHandler.java | 10 ++-----
.../ignite/internal/table/distributed/gc/MvGc.java | 4 +--
.../internal/table/distributed/IndexGcTest.java | 22 +++++++--------
.../gc/AbstractGcUpdateHandlerTest.java | 32 ++++++++++------------
.../internal/table/distributed/gc/MvGcTest.java | 12 ++++----
5 files changed, 34 insertions(+), 46 deletions(-)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
index 4fb0358565e..5b239514e23 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
@@ -68,11 +68,9 @@ public class GcUpdateHandler {
*
* @param lowWatermark Low watermark for the vacuum.
* @param count Count of entries to GC.
- * @param strict {@code true} if needed to remove the strictly passed {@code count} oldest stale entries, {@code false} if a premature
- * exit is allowed when it is not possible to acquire a lock for the {@link RowId}.
* @return {@code False} if there is no garbage left in the storage.
*/
- public boolean vacuumBatch(HybridTimestamp lowWatermark, int count, boolean strict) {
+ public boolean vacuumBatch(HybridTimestamp lowWatermark, int count) {
if (count <= 0) {
return true;
}
@@ -88,11 +86,7 @@ public class GcUpdateHandler {
case SUCCESS:
return true;
case FAILED_ACQUIRE_LOCK:
- if (strict) {
- continue;
- }
-
- return true;
+ continue;
case SHOULD_RELEASE:
// Storage engine needs resources (e.g., checkpoint needs write lock).
// Exit the closure to allow the engine to proceed.
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
index c8bd08da27d..b3857f92e14 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/MvGc.java
@@ -54,7 +54,7 @@ import org.jetbrains.annotations.TestOnly;
/**
* Garbage collector for multi-versioned storages and their indexes in the background.
*
- * @see GcUpdateHandler#vacuumBatch(HybridTimestamp, int, boolean)
+ * @see GcUpdateHandler#vacuumBatch
*/
public class MvGc implements ManuallyCloseable {
private static final IgniteLogger LOG = Loggers.forClass(MvGc.class);
@@ -252,7 +252,7 @@ public class MvGc implements ManuallyCloseable {
});
currentAwaitSafeTimeFuture
- .thenApplyAsync(unused -> gcUpdateHandler.vacuumBatch(lowWatermark, gcConfig.value().batchSize(), true), executor)
+ .thenApplyAsync(unused -> gcUpdateHandler.vacuumBatch(lowWatermark, gcConfig.value().batchSize()), executor)
.whenComplete((isGarbageLeft, throwable) -> {
if (throwable != null) {
if (hasCause(throwable, TrackerClosedException.class, StorageRemovedException.class)) {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
index 669b69bea8e..498f4edefe1 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexGcTest.java
@@ -42,7 +42,7 @@ public class IndexGcTest extends IndexBaseTest {
addWrite(storageUpdateHandler, rowUuid, row);
commitWrite(rowId);
- assertTrue(gcUpdateHandler.vacuumBatch(now(), 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(now(), 1));
assertEquals(1, getRowVersions(rowId).size());
// Newer entry has the same index value, so it should not be removed.
@@ -70,13 +70,13 @@ public class IndexGcTest extends IndexBaseTest {
HybridTimestamp afterCommits = now();
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
// row1 should still be in the index, because second write was identical to the first.
assertTrue(inAllIndexes(row1));
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
- assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
+ assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1));
assertEquals(1, getRowVersions(rowId).size());
// Older entries have different indexes, should be removed.
@@ -103,8 +103,8 @@ public class IndexGcTest extends IndexBaseTest {
HybridTimestamp afterCommits = now();
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
- assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
+ assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1));
assertEquals(0, getRowVersions(rowId).size());
// The last entry was a tombstone, so no indexes should be left.
@@ -129,8 +129,8 @@ public class IndexGcTest extends IndexBaseTest {
HybridTimestamp afterCommits = now();
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
- assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
+ assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1));
assertEquals(1, getRowVersions(rowId).size());
assertTrue(inAllIndexes(row));
@@ -154,9 +154,9 @@ public class IndexGcTest extends IndexBaseTest {
HybridTimestamp afterCommits = now();
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
- assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
- assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1, true));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
+ assertTrue(gcUpdateHandler.vacuumBatch(afterCommits, 1));
+ assertFalse(gcUpdateHandler.vacuumBatch(afterCommits, 1));
assertEquals(0, getRowVersions(rowId).size());
assertTrue(notInAnyIndex(row));
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
index c7bb5da96c7..0f8fcd8db21 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/AbstractGcUpdateHandlerTest.java
@@ -56,8 +56,6 @@ import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
import org.mockito.Mockito;
/**
@@ -82,9 +80,8 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
this.tableStorage = tableStorage;
}
- @ParameterizedTest(name = "strict : {0}")
- @ValueSource(booleans = {true, false})
- void testVacuum(boolean strict) {
+ @Test
+ void testVacuum() {
TestPartitionDataStorage partitionStorage = spy(createPartitionDataStorage());
IndexUpdateHandler indexUpdateHandler = spy(createIndexUpdateHandler());
@@ -92,7 +89,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
HybridTimestamp lowWatermark = HybridTimestamp.MAX_VALUE;
- assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 1, strict));
+ assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 1));
verify(partitionStorage).peek(lowWatermark);
// Let's check that StorageUpdateHandler#vacuumBatch returns true.
@@ -104,14 +101,13 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
addWriteCommitted(partitionStorage, rowId, row, clock.now());
addWriteCommitted(partitionStorage, rowId, row, clock.now());
- assertTrue(gcUpdateHandler.vacuumBatch(lowWatermark, 1, strict));
+ assertTrue(gcUpdateHandler.vacuumBatch(lowWatermark, 1));
verify(partitionStorage).peek(lowWatermark);
verify(indexUpdateHandler).tryRemoveFromIndexes(any(), eq(rowId), any(), isNull());
}
- @ParameterizedTest(name = "strict : {0}")
- @ValueSource(booleans = {true, false})
- void testVacuumBatch(boolean strict) {
+ @Test
+ void testVacuumBatch() {
TestPartitionDataStorage partitionStorage = spy(createPartitionDataStorage());
IndexUpdateHandler indexUpdateHandler = spy(createIndexUpdateHandler());
@@ -131,7 +127,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
addWriteCommitted(partitionStorage, rowId1, row1, clock.now());
addWriteCommitted(partitionStorage, rowId1, row1, clock.now());
- assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 5, strict));
+ assertFalse(gcUpdateHandler.vacuumBatch(lowWatermark, 5));
verify(partitionStorage, times(3)).peek(lowWatermark);
verify(indexUpdateHandler).tryRemoveFromIndexes(any(), eq(rowId0), any(), isNull());
@@ -162,8 +158,8 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
addWriteCommitted(partitionStorage, rowId1, null, clock.now());
runRace(
- () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 2, true),
- () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 2, true)
+ () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 2),
+ () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 2)
);
assertNull(partitionStorage.getStorage().closestRowId(RowId.lowestRowId(PARTITION_ID)));
@@ -199,8 +195,8 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
addWriteCommitted(partitionStorage, rowId3, null, clock.now());
runRace(
- () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 4, false),
- () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 4, false)
+ () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 4),
+ () -> gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, 4)
);
assertNull(partitionStorage.getStorage().closestRowId(RowId.lowestRowId(PARTITION_ID)));
@@ -243,7 +239,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
}
for (GcUpdateHandler gcUpdateHandler : gcUpdateHandlers) {
- gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, Integer.MAX_VALUE, true);
+ gcUpdateHandler.vacuumBatch(HybridTimestamp.MAX_VALUE, Integer.MAX_VALUE);
}
for (int i = 0; i < numPartitions; i++) {
@@ -284,7 +280,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
// When: Vacuum runs with shouldRelease() returning true (simulating lock contention)
when(shouldReleaseSupplier.getAsBoolean()).thenReturn(true);
- boolean hasGarbageLeft = gcUpdateHandler.vacuumBatch(lowWatermark, 10, true);
+ boolean hasGarbageLeft = gcUpdateHandler.vacuumBatch(lowWatermark, 10);
// Then: Vacuum exits early and reports garbage remaining
assertTrue(hasGarbageLeft, "Expected garbage to remain after early exit");
@@ -292,7 +288,7 @@ abstract class AbstractGcUpdateHandlerTest extends BaseMvStoragesTest {
// When: Vacuum runs again with shouldRelease() returning false
when(shouldReleaseSupplier.getAsBoolean()).thenReturn(false);
- hasGarbageLeft = gcUpdateHandler.vacuumBatch(lowWatermark, 100, true);
+ hasGarbageLeft = gcUpdateHandler.vacuumBatch(lowWatermark, 100);
// Then: All remaining garbage is processed
assertFalse(hasGarbageLeft, "Expected no garbage left after completing vacuum");
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
index ce4c665081d..3ed0edd984c 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/gc/MvGcTest.java
@@ -29,9 +29,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
@@ -389,7 +387,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
assertThat(startAwaitSafeTimeFuture, willCompleteSuccessfully());
assertThat(gc.removeStorage(tablePartitionId), willCompleteSuccessfully());
- verify(gcUpdateHandler, never()).vacuumBatch(any(), anyInt(), anyBoolean());
+ verify(gcUpdateHandler, never()).vacuumBatch(any(), anyInt());
}
private TablePartitionId createTablePartitionId() {
@@ -409,7 +407,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
CompletableFuture<Void> future,
@Nullable HybridTimestamp exp
) {
- when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
+ when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt())).then(invocation -> {
if (exp != null) {
try {
assertEquals(exp, invocation.getArgument(0));
@@ -429,7 +427,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
private static GcUpdateHandler createWithCountDownOnVacuum(CountDownLatch latch) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
- when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
+ when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt())).then(invocation -> {
latch.countDown();
return latch.getCount() > 0;
@@ -441,7 +439,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
private static GcUpdateHandler createWithWaitFinishVacuum(CompletableFuture<Void> startFuture, CompletableFuture<Void> finishFuture) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
- when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
+ when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt())).then(invocation -> {
startFuture.complete(null);
finishFuture.get(1, TimeUnit.SECONDS);
@@ -461,7 +459,7 @@ public class MvGcTest extends BaseIgniteAbstractTest {
private static GcUpdateHandler createWithCountDownOnVacuumWithoutNextBatch(CountDownLatch latch) {
GcUpdateHandler gcUpdateHandler = createGcUpdateHandler();
- when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt(), eq(true))).then(invocation -> {
+ when(gcUpdateHandler.vacuumBatch(any(HybridTimestamp.class), anyInt())).then(invocation -> {
latch.countDown();
// So that there is no processing of the next batch.
(ignite-3) 04/04: IGNITE-26998 wip
Posted by tk...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch ignite-26998
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit ba4127bc9634c4e79d57ae7698b384ef7df9a98d
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Fri Nov 28 12:06:48 2025 +0300
IGNITE-26998 wip
---
.../mv/AbstractPageMemoryMvPartitionStorage.java | 6 +-
.../table/distributed/gc/GcUpdateHandler.java | 70 ++++++++++------------
2 files changed, 36 insertions(+), 40 deletions(-)
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
index d2e9539901a..d0aeae565ac 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java
@@ -1094,7 +1094,11 @@ public abstract class AbstractPageMemoryMvPartitionStorage implements MvPartitio
} catch (IgniteInternalCheckedException e) {
throwStorageExceptionIfItCause(e);
- throw new StorageException("Row version lookup failed: [rowId={}, {}]", e, lowWatermark, createStorageInfo());
+ throw new StorageException(
+ "Peek GC entries failed: [lowWatermark={}, count={}, {}]",
+ e,
+ lowWatermark, count, createStorageInfo()
+ );
}
return res;
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
index c5ab546b188..341ddb84d90 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/gc/GcUpdateHandler.java
@@ -101,10 +101,16 @@ public class GcUpdateHandler {
}
private VacuumResult internalVacuumBatch(HybridTimestamp lowWatermark, IntHolder countHolder) {
- return storage.runConsistently(locker -> {
- int count = countHolder.get();
+ int count = countHolder.get();
+
+ List<GcEntry> peekEntries = storage.peek(lowWatermark, count);
- for (int i = 0; i < count; i++) {
+ if (peekEntries.isEmpty()) {
+ return VacuumResult.NO_GARBAGE_LEFT;
+ }
+
+ return storage.runConsistently(locker -> {
+ for (int i = 0; i < peekEntries.size(); i++) {
// Check if the storage engine needs resources before continuing.
if (locker.shouldRelease()) {
return VacuumResult.SHOULD_RELEASE;
@@ -112,9 +118,11 @@ public class GcUpdateHandler {
// It is safe for the first iteration to use a lock instead of tryLock, since there will be no deadlock for the first RowId
// and a deadlock may happen with subsequent ones.
- VacuumResult vacuumResult = internalVacuum(lowWatermark, locker, i > 0);
+ VacuumResult vacuumResult = internalVacuum(peekEntries.get(i), locker, i > 0);
- if (vacuumResult != VacuumResult.SUCCESS) {
+ if (vacuumResult == VacuumResult.REMOVED_BY_ANOTHER_THREAD) {
+ continue;
+ } else if (vacuumResult != VacuumResult.SUCCESS) {
return vacuumResult;
}
@@ -125,48 +133,32 @@ public class GcUpdateHandler {
});
}
- private VacuumResult internalVacuum(HybridTimestamp lowWatermark, Locker locker, boolean useTryLock) {
- while (true) {
- // Check if the storage engine needs resources before continuing.
- if (locker.shouldRelease()) {
- return VacuumResult.SHOULD_RELEASE;
- }
-
- // TODO: IGNITE-26998 Переделать а пока заглушка
- List<GcEntry> gcEntries = storage.peek(lowWatermark, 1);
+ private VacuumResult internalVacuum(GcEntry gcEntry, Locker locker, boolean useTryLock) {
+ RowId rowId = gcEntry.getRowId();
- if (gcEntries.isEmpty()) {
- return VacuumResult.NO_GARBAGE_LEFT;
+ if (useTryLock) {
+ if (!locker.tryLock(rowId)) {
+ return VacuumResult.FAILED_ACQUIRE_LOCK;
}
+ } else {
+ locker.lock(rowId);
+ }
- GcEntry gcEntry = gcEntries.get(0);
- RowId rowId = gcEntry.getRowId();
-
- if (useTryLock) {
- if (!locker.tryLock(rowId)) {
- return VacuumResult.FAILED_ACQUIRE_LOCK;
- }
- } else {
- locker.lock(rowId);
- }
-
- BinaryRow binaryRow = storage.vacuum(gcEntry);
-
- if (binaryRow == null) {
- // Removed by another thread, let's try to take another.
- continue;
- }
+ BinaryRow binaryRow = storage.vacuum(gcEntry);
- try (Cursor<ReadResult> cursor = storage.scanVersions(rowId)) {
- // TODO: IGNITE-21005 We need to choose only those indexes that are not available for transactions
- indexUpdateHandler.tryRemoveFromIndexes(binaryRow, rowId, cursor, null);
- }
+ if (binaryRow == null) {
+ return VacuumResult.REMOVED_BY_ANOTHER_THREAD;
+ }
- return VacuumResult.SUCCESS;
+ try (Cursor<ReadResult> cursor = storage.scanVersions(rowId)) {
+ // TODO: IGNITE-21005 We need to choose only those indexes that are not available for transactions
+ indexUpdateHandler.tryRemoveFromIndexes(binaryRow, rowId, cursor, null);
}
+
+ return VacuumResult.SUCCESS;
}
private enum VacuumResult {
- SUCCESS, NO_GARBAGE_LEFT, FAILED_ACQUIRE_LOCK, SHOULD_RELEASE
+ SUCCESS, NO_GARBAGE_LEFT, FAILED_ACQUIRE_LOCK, SHOULD_RELEASE, REMOVED_BY_ANOTHER_THREAD
}
}