This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new c454736a3b IGNITE-19290 Removing garbage from the partition, taking
into account the safe time (#1952)
c454736a3b is described below
commit c454736a3b361e135d7300c6c83f6069a3e2de1e
Author: Kirill Tkalenko <[email protected]>
AuthorDate: Tue May 2 13:32:45 2023 +0300
IGNITE-19290 Removing garbage from the partition, taking into account the
safe time (#1952)
---
.../ignite/distributed/ItTablePersistenceTest.java | 11 +-
.../distributed/ItTxDistributedTestSingleNode.java | 15 +-
.../internal/table/distributed/LowWatermark.java | 2 +-
.../table/distributed/StorageUpdateHandler.java | 96 +++-------
.../internal/table/distributed/TableManager.java | 8 +-
.../ignite/internal/table/distributed/gc/MvGc.java | 73 +++++---
.../AbstractMvStorageUpdateHandlerTest.java | 6 +-
.../internal/table/distributed/IndexBaseTest.java | 6 +-
.../PartitionGcOnWriteConcurrentTest.java | 167 -----------------
.../table/distributed/PartitionGcOnWriteTest.java | 197 ---------------------
.../distributed/StorageUpdateHandlerTest.java | 128 ++++++++++++-
.../internal/table/distributed/gc/MvGcTest.java | 50 +++++-
.../raft/PartitionCommandListenerTest.java | 11 +-
.../PartitionReplicaListenerIndexLockingTest.java | 9 +-
.../replication/PartitionReplicaListenerTest.java | 5 +-
.../table/impl/DummyInternalTableImpl.java | 10 +-
16 files changed, 314 insertions(+), 480 deletions(-)
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
index a996d2d73d..df4562b1c7 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
@@ -69,6 +69,7 @@ import
org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.rocksdb.RocksDbStorageEngine;
import
org.apache.ignite.internal.storage.rocksdb.configuration.schema.RocksDbStorageEngineConfiguration;
import org.apache.ignite.internal.table.InternalTable;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
@@ -376,18 +377,24 @@ public class ItTablePersistenceTest extends
ItAbstractListenerSnapshotTest<Parti
PartitionDataStorage partitionDataStorage = new
TestPartitionDataStorage(mvPartitionStorage);
+ PendingComparableValuesTracker<HybridTimestamp> safeTime =
new PendingComparableValuesTracker<>(
+ new HybridTimestamp(1, 0)
+ );
+
StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(
0,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
- zoneCfg.dataStorage()
+ zoneCfg.dataStorage(),
+ safeTime,
+ mock(LowWatermark.class)
);
PartitionListener listener = new PartitionListener(
partitionDataStorage,
storageUpdateHandler,
new TestTxStateStorage(),
- new PendingComparableValuesTracker<>(new
HybridTimestamp(1, 0)),
+ safeTime,
new PendingComparableValuesTracker<>(0L)
);
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
index 933dd1d29d..215d04fb91 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
@@ -27,6 +27,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.Mockito.mock;
import it.unimi.dsi.fastutil.ints.Int2ObjectOpenHashMap;
import java.util.ArrayList;
@@ -81,6 +82,7 @@ import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.TxAbstractTest;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessageGroup;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
@@ -123,7 +125,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mockito;
/**
* Distributed transaction test using a single partition table.
@@ -359,8 +360,8 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
1,
consistentIdToNode,
clientTxManager,
- Mockito.mock(MvTableStorage.class),
- Mockito.mock(TxStateTableStorage.class),
+ mock(MvTableStorage.class),
+ mock(TxStateTableStorage.class),
startClient() ? clientReplicaSvc :
replicaServices.get(localNodeName),
startClient() ? clientClock : clocks.get(localNodeName)
), new DummySchemaManagerImpl(ACCOUNTS_SCHEMA),
clientTxManager.lockManager());
@@ -372,8 +373,8 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
1,
consistentIdToNode,
clientTxManager,
- Mockito.mock(MvTableStorage.class),
- Mockito.mock(TxStateTableStorage.class),
+ mock(MvTableStorage.class),
+ mock(TxStateTableStorage.class),
startClient() ? clientReplicaSvc :
replicaServices.get(localNodeName),
startClient() ? clientClock : clocks.get(localNodeName)
), new DummySchemaManagerImpl(CUSTOMERS_SCHEMA),
clientTxManager.lockManager());
@@ -447,7 +448,9 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
partId,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
- dsCfg
+ dsCfg,
+ safeTime,
+ mock(LowWatermark.class)
);
TopologyAwareRaftGroupServiceFactory
topologyAwareRaftGroupServiceFactory = new TopologyAwareRaftGroupServiceFactory(
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/LowWatermark.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/LowWatermark.java
index 6fbf6609da..b84771e89a 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/LowWatermark.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/LowWatermark.java
@@ -148,7 +148,7 @@ public class LowWatermark implements ManuallyCloseable {
}
@Override
- public void close() throws Exception {
+ public void close() {
if (!closeGuard.compareAndSet(false, true)) {
return;
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
index 102ebacd3c..159859cf47 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
@@ -26,7 +26,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.BinaryRow;
@@ -41,6 +40,7 @@ import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.jetbrains.annotations.Nullable;
/**
@@ -55,12 +55,14 @@ public class StorageUpdateHandler {
private final TableIndexStoragesSupplier indexes;
- /** Last recorded GC low watermark. */
- private final AtomicReference<HybridTimestamp> lastRecordedLwm = new
AtomicReference<>();
-
- /** Data storage configuration. */
private final DataStorageConfiguration dsCfg;
+ /** Partition safe time tracker. */
+ private final PendingComparableValuesTracker<HybridTimestamp>
safeTimeTracker;
+
+ /** Low watermark. */
+ private final LowWatermark lowWatermark;
+
/**
* The constructor.
*
@@ -68,17 +70,22 @@ public class StorageUpdateHandler {
* @param storage Partition data storage.
* @param indexes Indexes supplier.
* @param dsCfg Data storage configuration.
+ * @param safeTimeTracker Partition safe time tracker.
*/
public StorageUpdateHandler(
int partitionId,
PartitionDataStorage storage,
TableIndexStoragesSupplier indexes,
- DataStorageConfiguration dsCfg
+ DataStorageConfiguration dsCfg,
+ PendingComparableValuesTracker<HybridTimestamp> safeTimeTracker,
+ LowWatermark lowWatermark
) {
this.partitionId = partitionId;
this.storage = storage;
this.indexes = indexes;
this.dsCfg = dsCfg;
+ this.safeTimeTracker = safeTimeTracker;
+ this.lowWatermark = lowWatermark;
}
/**
@@ -96,28 +103,6 @@ public class StorageUpdateHandler {
TablePartitionId commitPartitionId,
@Nullable ByteBuffer rowBuffer,
@Nullable Consumer<RowId> onReplication
- ) {
- handleUpdate(txId, rowUuid, commitPartitionId, rowBuffer,
onReplication, null);
- }
-
- /**
- * Handles single update.
- *
- * @param txId Transaction id.
- * @param rowUuid Row UUID.
- * @param commitPartitionId Commit partition id.
- * @param rowBuffer Row buffer.
- * @param onReplication Callback on replication.
- * @param lowWatermark GC low watermark.
- */
- // TODO: https://issues.apache.org/jira/browse/IGNITE-18909 Pass low
watermark.
- public void handleUpdate(
- UUID txId,
- UUID rowUuid,
- TablePartitionId commitPartitionId,
- @Nullable ByteBuffer rowBuffer,
- @Nullable Consumer<RowId> onReplication,
- @Nullable HybridTimestamp lowWatermark
) {
storage.runConsistently(() -> {
BinaryRow row = rowBuffer != null ? new ByteBufferRow(rowBuffer) :
null;
@@ -141,24 +126,7 @@ public class StorageUpdateHandler {
return null;
});
- executeBatchGc(lowWatermark);
- }
-
- /**
- * Handles multiple updates.
- *
- * @param txId Transaction id.
- * @param rowsToUpdate Collection of rows to update.
- * @param commitPartitionId Commit partition id.
- * @param onReplication On replication callback.
- */
- public void handleUpdateAll(
- UUID txId,
- Map<UUID, ByteBuffer> rowsToUpdate,
- TablePartitionId commitPartitionId,
- @Nullable Consumer<Collection<RowId>> onReplication
- ) {
- handleUpdateAll(txId, rowsToUpdate, commitPartitionId, onReplication,
null);
+ executeBatchGc();
}
/**
@@ -168,15 +136,12 @@ public class StorageUpdateHandler {
* @param rowsToUpdate Collection of rows to update.
* @param commitPartitionId Commit partition id.
* @param onReplication On replication callback.
- * @param lowWatermark GC low watermark.
*/
- // TODO: https://issues.apache.org/jira/browse/IGNITE-18909 Pass low
watermark.
public void handleUpdateAll(
UUID txId,
Map<UUID, ByteBuffer> rowsToUpdate,
TablePartitionId commitPartitionId,
- @Nullable Consumer<Collection<RowId>> onReplication,
- @Nullable HybridTimestamp lowWatermark
+ @Nullable Consumer<Collection<RowId>> onReplication
) {
storage.runConsistently(() -> {
UUID commitTblId = commitPartitionId.tableId();
@@ -208,29 +173,17 @@ public class StorageUpdateHandler {
return null;
});
- executeBatchGc(lowWatermark);
+ executeBatchGc();
}
- private void executeBatchGc(@Nullable HybridTimestamp newLwm) {
- if (newLwm == null) {
+ void executeBatchGc() {
+ HybridTimestamp lwm = lowWatermark.getLowWatermark();
+
+ if (lwm == null || safeTimeTracker.current().compareTo(lwm) < 0) {
return;
}
- @Nullable HybridTimestamp oldLwm;
- do {
- oldLwm = lastRecordedLwm.get();
-
- if (oldLwm != null && newLwm.compareTo(oldLwm) <= 0) {
- break;
- }
- } while (!lastRecordedLwm.compareAndSet(oldLwm, newLwm));
-
- if (oldLwm == null || newLwm.compareTo(oldLwm) > 0) {
- // Iff the lwm we have is the new lwm.
- // Otherwise our newLwm is either smaller than last recorded lwm
or last recorded lwm has changed
- // concurrently and it become greater. If that's the case, another
thread will perform the GC.
- vacuumBatch(newLwm, dsCfg.gcOnUpdateBatchSize().value());
- }
+ vacuumBatch(lwm, dsCfg.gcOnUpdateBatchSize().value());
}
/**
@@ -447,4 +400,11 @@ public class StorageUpdateHandler {
index.storage().setNextRowIdToBuild(finish ? null :
lastRowId.increment());
}
+
+ /**
+ * Returns the partition safe time tracker.
+ */
+ public PendingComparableValuesTracker<HybridTimestamp>
getSafeTimeTracker() {
+ return safeTimeTracker;
+ }
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index e5480730c3..c7a89bba12 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -742,7 +742,9 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
partId,
storage,
table.indexStorageAdapters(partId),
- dsCfg
+ dsCfg,
+ safeTimeTracker,
+ lowWatermark
);
mvGc.addStorage(replicaGrpId,
storageUpdateHandler);
@@ -2063,7 +2065,9 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
partId,
partitionDataStorage,
tbl.indexStorageAdapters(partId),
- dstZoneCfg.dataStorage()
+ dstZoneCfg.dataStorage(),
+ safeTimeTracker,
+ lowWatermark
);
RaftGroupOptions groupOptions =
groupOptionsForPartition(
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 a29f4e6d36..4686233c18 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
@@ -41,6 +41,7 @@ import
org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.internal.util.TrackerClosedException;
import org.apache.ignite.lang.ErrorGroups.GarbageCollector;
import org.apache.ignite.lang.IgniteInternalException;
import org.jetbrains.annotations.TestOnly;
@@ -194,8 +195,8 @@ public class MvGc implements ManuallyCloseable {
}
private void scheduleGcForStorage(TablePartitionId tablePartitionId) {
- executor.submit(() -> inBusyLock(() -> {
- CompletableFuture<Void> future = new CompletableFuture<>();
+ inBusyLock(() -> {
+ CompletableFuture<Void> currentGcFuture = new
CompletableFuture<>();
GcStorageHandler storageHandler =
storageHandlerByPartitionId.compute(tablePartitionId, (tablePartId,
gcStorageHandler) -> {
if (gcStorageHandler == null) {
@@ -206,7 +207,7 @@ public class MvGc implements ManuallyCloseable {
CompletableFuture<Void> inProgressFuture =
gcStorageHandler.gcInProgressFuture.get();
if (inProgressFuture == null || inProgressFuture.isDone()) {
- boolean casResult =
gcStorageHandler.gcInProgressFuture.compareAndSet(inProgressFuture, future);
+ boolean casResult =
gcStorageHandler.gcInProgressFuture.compareAndSet(inProgressFuture,
currentGcFuture);
assert casResult : tablePartId;
} else {
@@ -221,35 +222,63 @@ public class MvGc implements ManuallyCloseable {
return;
}
- if (storageHandler.gcInProgressFuture.get() != future) {
+ if (storageHandler.gcInProgressFuture.get() != currentGcFuture) {
// Someone in parallel is already collecting garbage, we will
try once again after completion of gcInProgressFuture.
return;
}
try {
- for (int i = 0; i < GC_BATCH_SIZE; i++) {
- HybridTimestamp lowWatermark = lowWatermarkReference.get();
+ HybridTimestamp lowWatermark = lowWatermarkReference.get();
- assert lowWatermark != null : tablePartitionId;
+ assert lowWatermark != null : tablePartitionId;
- // If storage has been deleted or there is no garbage,
then for now we will stop collecting garbage for this storage.
- if
(!storageHandlerByPartitionId.containsKey(tablePartitionId)
- ||
!storageHandler.storageUpdateHandler.vacuum(lowWatermark)) {
- return;
- }
- }
- } catch (Throwable t) {
- future.completeExceptionally(t);
+ // If the storage has been deleted, then garbage collection is
no longer necessary.
+ if
(!storageHandlerByPartitionId.containsKey(tablePartitionId)) {
+ currentGcFuture.complete(null);
- return;
- } finally {
- if (!future.isCompletedExceptionally()) {
- future.complete(null);
+ return;
}
- }
- scheduleGcForStorage(tablePartitionId);
- }));
+ StorageUpdateHandler storageUpdateHandler =
storageHandler.storageUpdateHandler;
+
+ // We can only start garbage collection when the partition
safe time is reached.
+ storageUpdateHandler.getSafeTimeTracker()
+ .waitFor(lowWatermark)
+ .thenApplyAsync(unused -> {
+ for (int i = 0; i < GC_BATCH_SIZE; i++) {
+ // If the storage has been deleted or there is
no garbage, then we will stop.
+ if
(!storageHandlerByPartitionId.containsKey(tablePartitionId)
+ ||
!storageUpdateHandler.vacuum(lowWatermark)) {
+ return false;
+ }
+ }
+
+ return true;
+ }, executor)
+ .whenComplete((isGarbageLeft, throwable) -> {
+ if (throwable != null) {
+ if (throwable instanceof TrackerClosedException
+ || throwable.getCause() instanceof
TrackerClosedException) {
+ currentGcFuture.complete(null);
+ } else {
+
currentGcFuture.completeExceptionally(throwable);
+ }
+
+ return;
+ }
+
+ currentGcFuture.complete(null);
+
+ // If there is garbage left and the storage has
not been deleted, then we will schedule the next garbage
+ // collection.
+ if (isGarbageLeft &&
storageHandlerByPartitionId.containsKey(tablePartitionId)) {
+ scheduleGcForStorage(tablePartitionId);
+ }
+ });
+ } catch (Throwable t) {
+ currentGcFuture.completeExceptionally(t);
+ }
+ });
}
private <T> T inBusyLock(Supplier<T> supplier) {
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/AbstractMvStorageUpdateHandlerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/AbstractMvStorageUpdateHandlerTest.java
index 8f060b55fe..dbae8274c0 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/AbstractMvStorageUpdateHandlerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/AbstractMvStorageUpdateHandlerTest.java
@@ -21,6 +21,7 @@ import static
org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
import java.util.Map;
import java.util.UUID;
@@ -40,6 +41,7 @@ import
org.apache.ignite.internal.storage.engine.StorageEngine;
import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -95,7 +97,9 @@ abstract class AbstractMvStorageUpdateHandlerTest extends
BaseMvStoragesTest {
PARTITION_ID,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
- distributionZoneConfig.dataStorage()
+ distributionZoneConfig.dataStorage(),
+ new
PendingComparableValuesTracker<>(HybridTimestamp.MAX_VALUE),
+ mock(LowWatermark.class)
);
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
index fbb24b636c..66cce1a5c0 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.table.distributed;
import static java.util.Collections.singletonMap;
import static java.util.stream.Collectors.toList;
+import static org.mockito.Mockito.mock;
import java.util.List;
import java.util.Map;
@@ -46,6 +47,7 @@ import
org.apache.ignite.internal.storage.index.impl.TestSortedIndexStorage;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -129,7 +131,9 @@ public abstract class IndexBaseTest extends
BaseMvStoragesTest {
PARTITION_ID,
new TestPartitionDataStorage(storage),
DummyInternalTableImpl.createTableIndexStoragesSupplier(indexes),
- dsCfg
+ dsCfg,
+ new
PendingComparableValuesTracker<>(HybridTimestamp.MAX_VALUE),
+ mock(LowWatermark.class)
);
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
deleted file mode 100644
index c782bb327e..0000000000
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
+++ /dev/null
@@ -1,167 +0,0 @@
-/*
- * 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.table.distributed;
-
-import static org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.atMostOnce;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import java.nio.ByteBuffer;
-import java.util.Collections;
-import java.util.Map;
-import java.util.UUID;
-import java.util.function.Consumer;
-import org.apache.ignite.distributed.TestPartitionDataStorage;
-import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
-import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.hlc.HybridClock;
-import org.apache.ignite.internal.hlc.HybridClockImpl;
-import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.schema.ByteBufferRow;
-import
org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration;
-import org.apache.ignite.internal.storage.MvPartitionStorage;
-import org.apache.ignite.internal.storage.MvPartitionStorage.WriteClosure;
-import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
-import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
-import org.apache.ignite.internal.testframework.IgniteTestUtils.RunnableX;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.EnumSource;
-
-/** Tests for concurrent cooperative GC (GC that is executed on write). */
-@ExtendWith(ConfigurationExtension.class)
-public class PartitionGcOnWriteConcurrentTest {
- private static final int PARTITION_ID = 1;
- private static final TablePartitionId TABLE_PARTITION_ID = new
TablePartitionId(UUID.randomUUID(), PARTITION_ID);
- private static final HybridClock CLOCK = new HybridClockImpl();
-
- private MvPartitionStorage storage;
- private StorageUpdateHandler storageUpdateHandler;
-
- @BeforeEach
- void setUp(@InjectConfiguration DataStorageConfiguration dsCfg) {
- storage = mock(MvPartitionStorage.class);
- doAnswer(invocation -> {
- WriteClosure<?> cls = invocation.getArgument(0);
-
- return cls.execute();
- }).when(storage).runConsistently(any());
-
- when(storage.pollForVacuum(any())).thenReturn(null);
-
- storageUpdateHandler = new StorageUpdateHandler(
- PARTITION_ID,
- new TestPartitionDataStorage(storage),
-
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
- dsCfg
- );
- }
-
- @ParameterizedTest
- @EnumSource(UpdateType.class)
- void testSameLwm(UpdateType updateType) {
- HybridTimestamp lwm = CLOCK.now();
-
- runRace(createRunnable(updateType, lwm), createRunnable(updateType,
lwm));
-
- verify(storage, times(1)).pollForVacuum(lwm);
- }
-
- @ParameterizedTest
- @EnumSource(UpdateType.class)
- void testDifferentLwm(UpdateType updateType) {
- int count = 10;
-
- HybridTimestamp[] timestamps = new HybridTimestamp[count];
-
- RunnableX[] runnables = new RunnableX[count];
-
- for (int i = 0; i < count; i++) {
- HybridTimestamp ts = CLOCK.now();
-
- timestamps[i] = ts;
-
- runnables[i] = createRunnable(updateType, ts);
- }
-
- runRace(runnables);
-
- for (int i = 0; i < count - 1; i++) {
- verify(storage, atMostOnce()).pollForVacuum(timestamps[i]);
- }
-
- verify(storage, times(1)).pollForVacuum(timestamps[count - 1]);
- }
-
- @ParameterizedTest
- @EnumSource(UpdateType.class)
- void testDifferentLwmWithPreviousVacuums(UpdateType updateType) throws
Throwable {
- HybridTimestamp lwm1 = CLOCK.now();
- HybridTimestamp lwm2 = CLOCK.now();
- HybridTimestamp lwm3 = CLOCK.now();
-
- createRunnable(updateType, lwm1).run();
-
- runRace(createRunnable(updateType, lwm2), createRunnable(updateType,
lwm3));
-
- verify(storage, times(1)).pollForVacuum(lwm1);
- verify(storage, atMostOnce()).pollForVacuum(lwm2);
- verify(storage, times(1)).pollForVacuum(lwm3);
- }
-
- private RunnableX createRunnable(UpdateType updateType, HybridTimestamp
lwm) {
- if (updateType == UpdateType.UPDATE) {
- //noinspection unchecked
- return () -> storageUpdateHandler.handleUpdate(
- UUID.randomUUID(),
- UUID.randomUUID(),
- TABLE_PARTITION_ID,
- buffer(),
- mock(Consumer.class),
- lwm
- );
- } else {
- //noinspection unchecked
- return () -> storageUpdateHandler.handleUpdateAll(
- UUID.randomUUID(),
- Collections.emptyMap(),
- TABLE_PARTITION_ID,
- mock(Consumer.class),
- lwm
- );
- }
- }
-
- private static ByteBuffer buffer() {
- ByteBuffer buf = mock(ByteBuffer.class);
- when(buf.order()).thenReturn(ByteBufferRow.ORDER);
-
- return buf;
- }
-
- private enum UpdateType {
- UPDATE,
- UPDATE_ALL;
- }
-}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
deleted file mode 100644
index 39721d97b1..0000000000
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
+++ /dev/null
@@ -1,197 +0,0 @@
-/*
- * 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.table.distributed;
-
-import static java.util.Collections.singletonMap;
-import static java.util.stream.Collectors.toList;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import org.apache.ignite.distributed.TestPartitionDataStorage;
-import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
-import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
-import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.schema.BinaryRow;
-import
org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration;
-import org.apache.ignite.internal.storage.BaseMvStoragesTest;
-import org.apache.ignite.internal.storage.ReadResult;
-import org.apache.ignite.internal.storage.RowId;
-import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
-import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
-import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
-import org.apache.ignite.internal.util.Cursor;
-import org.jetbrains.annotations.Nullable;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.EnumSource;
-
-/** Tests for cooperative GC (GC that is executed on write). */
-@ExtendWith(ConfigurationExtension.class)
-public class PartitionGcOnWriteTest extends BaseMvStoragesTest {
- private static final int WRITES_COUNT = 10;
- private static final int GC_BATCH_SIZE = 7;
- private static final UUID TX_ID = UUID.randomUUID();
- private static final int PARTITION_ID = 1;
- private static final TablePartitionId TABLE_PARTITION_ID = new
TablePartitionId(UUID.randomUUID(), PARTITION_ID);
-
- private TestMvPartitionStorage storage;
- private StorageUpdateHandler storageUpdateHandler;
-
- @BeforeEach
- void setUp(@InjectConfiguration("mock.gcOnUpdateBatchSize=" +
GC_BATCH_SIZE) DataStorageConfiguration dsCfg) {
- storage = new TestMvPartitionStorage(1);
-
- storageUpdateHandler = new StorageUpdateHandler(
- 1,
- new TestPartitionDataStorage(storage),
-
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
- dsCfg
- );
- }
-
- @ParameterizedTest
- @EnumSource(AddWriteWithGc.class)
- void testNullLwm(AddWriteWithGc writer) {
- RowId rowId = fillWithDataForGc();
-
- writeWithGc(writer, null);
-
- assertEquals(WRITES_COUNT, getRowVersions(rowId).size());
- }
-
- @ParameterizedTest
- @EnumSource(AddWriteWithGc.class)
- void testOlderLwm(AddWriteWithGc writer) {
- HybridTimestamp older = clock.now();
-
- HybridTimestamp newer = clock.now();
-
- writeWithGc(writer, newer);
-
- RowId rowId = fillWithDataForGc();
-
- writeWithGc(writer, older);
-
- assertEquals(WRITES_COUNT, getRowVersions(rowId).size());
- }
-
- @ParameterizedTest
- @EnumSource(AddWriteWithGc.class)
- void testNewerLwm(AddWriteWithGc writer) {
- RowId rowId = fillWithDataForGc();
-
- writeWithGc(writer, clock.now());
-
- assertEquals(WRITES_COUNT - GC_BATCH_SIZE,
getRowVersions(rowId).size());
- }
-
- private List<ReadResult> getRowVersions(RowId rowId) {
- try (Cursor<ReadResult> readResults = storage.scanVersions(rowId)) {
- return readResults.stream().collect(toList());
- }
- }
-
- private void writeWithGc(AddWriteWithGc writer, @Nullable HybridTimestamp
lwm) {
- UUID rowUuid = UUID.randomUUID();
-
- TestKey key = new TestKey(1337, "leet");
- BinaryRow row = binaryRow(key, new TestValue(999, "bar"));
-
- writer.addWrite(storageUpdateHandler, rowUuid, row, lwm);
- }
-
- private RowId fillWithDataForGc() {
- UUID rowUuid = UUID.randomUUID();
- var rowId = new RowId(PARTITION_ID, rowUuid);
-
- TestKey key = new TestKey(1, "foo");
-
- for (int i = 0; i < WRITES_COUNT; i++) {
- BinaryRow row = binaryRow(key, new TestValue(i, "bar" + i));
-
- addWrite(storageUpdateHandler, rowUuid, row);
-
- commitWrite(rowId);
- }
-
- return rowId;
- }
-
- private static void addWrite(StorageUpdateHandler handler, UUID rowUuid,
@Nullable BinaryRow row) {
- handler.handleUpdate(
- TX_ID,
- rowUuid,
- TABLE_PARTITION_ID,
- row == null ? null : row.byteBuffer(),
- (unused) -> {}
- );
- }
-
- private void commitWrite(RowId rowId) {
- storage.runConsistently(() -> {
- storage.commitWrite(rowId, clock.now());
-
- return null;
- });
- }
-
- /** Enum that encapsulates update API. */
- enum AddWriteWithGc {
- /** Uses update api. */
- USE_UPDATE {
- @Override
- void addWrite(StorageUpdateHandler handler, TablePartitionId
partitionId, UUID rowUuid, @Nullable BinaryRow row,
- @Nullable HybridTimestamp lwm) {
- handler.handleUpdate(
- TX_ID,
- rowUuid,
- partitionId,
- row == null ? null : row.byteBuffer(),
- (unused) -> {},
- lwm
- );
- }
- },
- /** Uses updateAll api. */
- USE_UPDATE_ALL {
- @Override
- void addWrite(StorageUpdateHandler handler, TablePartitionId
partitionId, UUID rowUuid, @Nullable BinaryRow row,
- @Nullable HybridTimestamp lwm) {
- handler.handleUpdateAll(
- TX_ID,
- singletonMap(rowUuid, row == null ? null :
row.byteBuffer()),
- partitionId,
- (unused) -> {},
- lwm
- );
- }
- };
-
- void addWrite(StorageUpdateHandler handler, UUID rowUuid, @Nullable
BinaryRow row, @Nullable HybridTimestamp lwm) {
- TablePartitionId tablePartitionId = new
TablePartitionId(UUID.randomUUID(), 1);
-
- addWrite(handler, tablePartitionId, rowUuid, row, lwm);
- }
-
- abstract void addWrite(StorageUpdateHandler handler, TablePartitionId
partitionId, UUID rowUuid, @Nullable BinaryRow row,
- @Nullable HybridTimestamp lwm);
- }
-}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
index 69590f7603..59d43128cf 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
@@ -20,8 +20,13 @@ package org.apache.ignite.internal.table.distributed;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
import static java.util.stream.Collectors.toList;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -30,17 +35,25 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.distributed.TestPartitionDataStorage;
import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.BinaryRow;
import
org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration;
+import org.apache.ignite.internal.storage.BinaryRowAndRowId;
import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import org.apache.ignite.internal.storage.index.IndexStorage;
import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
+import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.internal.util.CursorUtils;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -56,6 +69,12 @@ public class StorageUpdateHandlerTest {
private final HybridClock clock = new HybridClockImpl();
+ private final PendingComparableValuesTracker<HybridTimestamp>
safeTimeTracker = spy(new PendingComparableValuesTracker<>(
+ new HybridTimestamp(1, 0)
+ ));
+
+ private final LowWatermark lowWatermark = mock(LowWatermark.class);
+
@Test
void testBuildIndex() {
PartitionDataStorage partitionStorage =
mock(PartitionDataStorage.class);
@@ -104,6 +123,107 @@ public class StorageUpdateHandlerTest {
verify(indexes, times(2)).addIndexToWaitIfAbsent(indexId);
}
+ @Test
+ void testVacuum() {
+ PartitionDataStorage partitionStorage = createPartitionDataStorage();
+
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler(partitionStorage,
mock(TableIndexStoragesSupplier.class));
+
+ HybridTimestamp lowWatermark = new HybridTimestamp(100, 100);
+
+ assertFalse(storageUpdateHandler.vacuum(lowWatermark));
+ verify(partitionStorage).pollForVacuum(lowWatermark);
+ // Let's check that StorageUpdateHandler#vacuumBatch returns true.
+ clearInvocations(partitionStorage);
+
+ BinaryRowAndRowId binaryRowAndRowId = new
BinaryRowAndRowId(mock(BinaryRow.class), new RowId(PARTITION_ID));
+
+
when(partitionStorage.scanVersions(any(RowId.class))).thenReturn(CursorUtils.emptyCursor());
+
when(partitionStorage.pollForVacuum(lowWatermark)).thenReturn(binaryRowAndRowId);
+
+ assertTrue(storageUpdateHandler.vacuum(lowWatermark));
+ verify(partitionStorage).pollForVacuum(lowWatermark);
+ }
+
+ @Test
+ void testExecuteBatchGc() {
+ PartitionDataStorage partitionStorage = createPartitionDataStorage();
+
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler(partitionStorage,
mock(TableIndexStoragesSupplier.class));
+
+ AtomicReference<HybridTimestamp> lowWatermarkReference = new
AtomicReference<>();
+
+ when(lowWatermark.getLowWatermark()).then(invocation ->
lowWatermarkReference.get());
+
+ // Let's check that if lwm is {@code null} then nothing will happen.
+ storageUpdateHandler.executeBatchGc();
+
+ verify(partitionStorage,
never()).pollForVacuum(any(HybridTimestamp.class));
+
+ // Let's check that if lvm is greater than the safe time, then nothing
will happen.
+ safeTimeTracker.update(new HybridTimestamp(10, 10));
+
+ lowWatermarkReference.set(new HybridTimestamp(11, 1));
+
+ storageUpdateHandler.executeBatchGc();
+
+ verify(partitionStorage,
never()).pollForVacuum(any(HybridTimestamp.class));
+
+ // Let's check that if lvm is equal to or less than the safe time,
then garbage collection will be executed.
+ lowWatermarkReference.set(new HybridTimestamp(10, 10));
+
+ storageUpdateHandler.executeBatchGc();
+
+ verify(partitionStorage,
times(1)).pollForVacuum(any(HybridTimestamp.class));
+
+ lowWatermarkReference.set(new HybridTimestamp(9, 9));
+
+ storageUpdateHandler.executeBatchGc();
+
+ verify(partitionStorage,
times(2)).pollForVacuum(any(HybridTimestamp.class));
+ }
+
+ @Test
+ void testInvokeGcOnHandleUpdate() {
+ PartitionDataStorage partitionStorage = createPartitionDataStorage();
+
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler(partitionStorage,
mock(TableIndexStoragesSupplier.class));
+
+ HybridTimestamp lwm = safeTimeTracker.current();
+
+ when(lowWatermark.getLowWatermark()).thenReturn(lwm);
+
+ storageUpdateHandler.handleUpdate(
+ UUID.randomUUID(),
+ UUID.randomUUID(),
+ new TablePartitionId(UUID.randomUUID(), PARTITION_ID),
+ null,
+ null
+ );
+
+ verify(partitionStorage).pollForVacuum(lwm);
+ }
+
+ @Test
+ void testInvokeGcOnHandleUpdateAll() {
+ PartitionDataStorage partitionStorage = createPartitionDataStorage();
+
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler(partitionStorage,
mock(TableIndexStoragesSupplier.class));
+
+ HybridTimestamp lwm = safeTimeTracker.current();
+
+ when(lowWatermark.getLowWatermark()).thenReturn(lwm);
+
+ storageUpdateHandler.handleUpdateAll(
+ UUID.randomUUID(),
+ Map.of(),
+ new TablePartitionId(UUID.randomUUID(), PARTITION_ID),
+ null
+ );
+
+ verify(partitionStorage).pollForVacuum(lwm);
+ }
+
private static TableSchemaAwareIndexStorage createIndexStorage() {
TableSchemaAwareIndexStorage indexStorage =
mock(TableSchemaAwareIndexStorage.class);
@@ -115,7 +235,7 @@ public class StorageUpdateHandlerTest {
}
private StorageUpdateHandler
createStorageUpdateHandler(PartitionDataStorage partitionStorage,
TableIndexStoragesSupplier indexes) {
- return new StorageUpdateHandler(PARTITION_ID, partitionStorage,
indexes, dataStorageConfig);
+ return new StorageUpdateHandler(PARTITION_ID, partitionStorage,
indexes, dataStorageConfig, safeTimeTracker, lowWatermark);
}
private void setRowVersions(PartitionDataStorage partitionStorage,
Map<UUID, List<BinaryRow>> rowVersions) {
@@ -129,4 +249,10 @@ public class StorageUpdateHandlerTest {
when(partitionStorage.scanVersions(rowId)).thenReturn(Cursor.fromIterable(readResults));
}
}
+
+ private static PartitionDataStorage createPartitionDataStorage() {
+ PartitionDataStorage partitionStorage = spy(new
TestPartitionDataStorage(new TestMvPartitionStorage(PARTITION_ID)));
+
+ return partitionStorage;
+ }
}
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 2d7b56a2da..56caf6c180 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
@@ -21,6 +21,7 @@ import static
org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowFast;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willTimeoutFast;
import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.apache.ignite.internal.util.IgniteUtils.closeAllManually;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
@@ -29,6 +30,8 @@ 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.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.UUID;
@@ -41,6 +44,7 @@ import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.lang.ErrorGroups.GarbageCollector;
import org.apache.ignite.lang.IgniteInternalException;
import org.jetbrains.annotations.Nullable;
@@ -291,7 +295,7 @@ public class MvGcTest {
void testClose() throws Exception {
gc.close();
- assertThrowsClosed(() -> gc.addStorage(createTablePartitionId(),
mock(StorageUpdateHandler.class)));
+ assertThrowsClosed(() -> gc.addStorage(createTablePartitionId(),
createStorageUpdateHandler()));
assertThrowsClosed(() -> gc.removeStorage(createTablePartitionId()));
assertThrowsClosed(() -> gc.updateLowWatermark(new HybridTimestamp(1,
1)));
@@ -335,12 +339,40 @@ public class MvGcTest {
}
}
+ @Test
+ void testInvokeVacuumOnlyAfterReachSafeTime() {
+ CompletableFuture<Void> invokeVacuumMethodFuture = new
CompletableFuture<>();
+
+ HybridTimestamp lvm = new HybridTimestamp(10, 10);
+
+ StorageUpdateHandler storageUpdateHandler =
createWithCompleteFutureOnVacuum(invokeVacuumMethodFuture, lvm);
+
+ PendingComparableValuesTracker<HybridTimestamp> safeTimeTracker =
spy(new PendingComparableValuesTracker<>(
+ HybridTimestamp.MIN_VALUE
+ ));
+
+
when(storageUpdateHandler.getSafeTimeTracker()).thenReturn(safeTimeTracker);
+
+ gc.addStorage(createTablePartitionId(), storageUpdateHandler);
+
+ // Let's update the low watermark and see that we didn't start the
garbage collection because we didn't reach the safe time.
+ gc.updateLowWatermark(lvm);
+
+ assertThat(invokeVacuumMethodFuture, willTimeoutFast());
+ verify(safeTimeTracker).waitFor(lvm);
+
+ // Update the safe time to be equal to the low watermark and make sure
the garbage collection starts.
+ safeTimeTracker.update(lvm);
+
+ assertThat(invokeVacuumMethodFuture, willSucceedFast());
+ }
+
private TablePartitionId createTablePartitionId() {
return new TablePartitionId(UUID.randomUUID(), PARTITION_ID);
}
private StorageUpdateHandler
createWithCompleteFutureOnVacuum(CompletableFuture<Void> future, @Nullable
HybridTimestamp exp) {
- StorageUpdateHandler storageUpdateHandler =
mock(StorageUpdateHandler.class);
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler();
completeFutureOnVacuum(storageUpdateHandler, future, exp);
@@ -370,7 +402,7 @@ public class MvGcTest {
}
private StorageUpdateHandler createWithCountDownOnVacuum(CountDownLatch
latch) {
- StorageUpdateHandler storageUpdateHandler =
mock(StorageUpdateHandler.class);
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler();
when(storageUpdateHandler.vacuum(any(HybridTimestamp.class))).then(invocation
-> {
latch.countDown();
@@ -382,7 +414,7 @@ public class MvGcTest {
}
private StorageUpdateHandler
createWithWaitFinishVacuum(CompletableFuture<Void> startFuture,
CompletableFuture<Void> finishFuture) {
- StorageUpdateHandler storageUpdateHandler =
mock(StorageUpdateHandler.class);
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler();
when(storageUpdateHandler.vacuum(any(HybridTimestamp.class))).then(invocation
-> {
startFuture.complete(null);
@@ -402,7 +434,7 @@ public class MvGcTest {
}
private StorageUpdateHandler
createWithCountDownOnVacuumWithoutNextBatch(CountDownLatch latch) {
- StorageUpdateHandler storageUpdateHandler =
mock(StorageUpdateHandler.class);
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler();
when(storageUpdateHandler.vacuum(any(HybridTimestamp.class))).then(invocation
-> {
latch.countDown();
@@ -413,4 +445,12 @@ public class MvGcTest {
return storageUpdateHandler;
}
+
+ private static StorageUpdateHandler createStorageUpdateHandler() {
+ StorageUpdateHandler storageUpdateHandler =
mock(StorageUpdateHandler.class);
+
+ when(storageUpdateHandler.getSafeTimeTracker()).thenReturn(new
PendingComparableValuesTracker<>(HybridTimestamp.MAX_VALUE));
+
+ return storageUpdateHandler;
+ }
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
index 00df7072a6..d99ca36a01 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/raft/PartitionCommandListenerTest.java
@@ -80,6 +80,7 @@ import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
@@ -189,7 +190,9 @@ public class PartitionCommandListenerTest {
PARTITION_ID,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.id(),
pkStorage)),
- dsCfg
+ dsCfg,
+ safeTimeTracker,
+ mock(LowWatermark.class)
));
commandListener = new PartitionListener(
@@ -283,14 +286,16 @@ public class PartitionCommandListenerTest {
PARTITION_ID,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.id(),
pkStorage)),
- dsCfg
+ dsCfg,
+ safeTimeTracker,
+ mock(LowWatermark.class)
);
PartitionListener testCommandListener = new PartitionListener(
partitionDataStorage,
storageUpdateHandler,
txStateStorage,
- new PendingComparableValuesTracker<>(new HybridTimestamp(1,
0)),
+ safeTimeTracker,
new PendingComparableValuesTracker<>(0L)
);
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index 9b53b6a251..3e39392828 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -61,6 +61,7 @@ import
org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.storage.index.impl.TestSortedIndexStorage;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.SortedIndexLocker;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
@@ -165,6 +166,8 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
);
DummySchemaManagerImpl schemaManager = new
DummySchemaManagerImpl(schemaDescriptor);
+ PendingComparableValuesTracker<HybridTimestamp> safeTime = new
PendingComparableValuesTracker<>(CLOCK.now());
+
partitionReplicaListener = new PartitionReplicaListener(
TEST_MV_PARTITION_STORAGE,
mockRaftClient,
@@ -184,14 +187,16 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
hashIndexLocker.id(), hashIndexStorage
),
CLOCK,
- new PendingComparableValuesTracker<>(CLOCK.now()),
+ safeTime,
new TestTxStateStorage(),
mock(PlacementDriver.class),
new StorageUpdateHandler(
PART_ID,
new
TestPartitionDataStorage(TEST_MV_PARTITION_STORAGE),
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
- dsCfg
+ dsCfg,
+ safeTime,
+ mock(LowWatermark.class)
),
peer -> true,
CompletableFuture.completedFuture(schemaManager)
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 2c64dcd142..ffb75897bb 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -89,6 +89,7 @@ import
org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.storage.index.impl.TestSortedIndexStorage;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.SortedIndexLocker;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
@@ -338,7 +339,9 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
partId,
partitionDataStorage,
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
- dsCfg
+ dsCfg,
+ safeTimeClock,
+ mock(LowWatermark.class)
),
peer -> localNode.name().equals(peer.consistentId()),
completedFuture(schemaManager)
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index b05940b7a4..e4da2afd8a 100644
---
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -61,6 +61,7 @@ import
org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
+import org.apache.ignite.internal.table.distributed.LowWatermark;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableIndexStoragesSupplier;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
@@ -274,7 +275,14 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
lenient().when(gcBatchSizeValue.value()).thenReturn(5);
lenient().when(dsCfg.gcOnUpdateBatchSize()).thenReturn(gcBatchSizeValue);
- StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(PART_ID, partitionDataStorage, indexes, dsCfg);
+ StorageUpdateHandler storageUpdateHandler = new StorageUpdateHandler(
+ PART_ID,
+ partitionDataStorage,
+ indexes,
+ dsCfg,
+ safeTime,
+ mock(LowWatermark.class)
+ );
DummySchemaManagerImpl schemaManager = new
DummySchemaManagerImpl(schema);
