You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2022/07/27 08:41:03 UTC
[ignite-3] branch main updated: IGNITE-16907 Init RAFT group with last applied index that corresponds to persisted storage data. (#937)
This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 d7194b4e9 IGNITE-16907 Init RAFT group with last applied index that corresponds to persisted storage data. (#937)
d7194b4e9 is described below
commit d7194b4e911ef79cc61cbfad4e6dbe705a458846
Author: ibessonov <be...@gmail.com>
AuthorDate: Wed Jul 27 11:40:59 2022 +0300
IGNITE-16907 Init RAFT group with last applied index that corresponds to persisted storage data. (#937)
---
.../ignite/client/fakes/FakeInternalTable.java | 13 +-
.../ignite/raft/client/service/CommandClosure.java | 8 ++
.../apache/ignite/internal/raft/ItLozaTest.java | 11 +-
.../java/org/apache/ignite/internal/raft/Loza.java | 134 +++++++-------------
.../internal/raft/server/impl/JraftServerImpl.java | 24 +++-
.../storage/snapshot/SnapshotExecutorImpl.java | 8 +-
.../org/apache/ignite/internal/raft/LozaTest.java | 2 +-
.../internal/raft/server/impl/RaftServerImpl.java | 14 +++
.../sql/engine/schema/IgniteTableImpl.java | 4 +-
.../internal/sql/engine/StopCalciteModuleTest.java | 4 +-
.../sql/engine/exec/MockedStructuresTest.java | 2 +-
.../internal/storage/MvPartitionStorage.java | 37 ++++++
.../ignite/internal/storage/PartitionStorage.java | 3 +
.../internal/storage/engine/MvTableStorage.java | 5 +
.../internal/storage/engine/TableStorage.java | 3 +
.../storage/basic/TestMvPartitionStorage.java | 48 ++++++++
.../internal/storage/chm/TestMvTableStorage.java | 8 +-
.../VolatilePageMemoryStorageEngine.java | 6 +
.../mv/PageMemoryMvPartitionStorage.java | 40 ++++++
.../storage/rocksdb/RocksDbMetaStorage.java | 19 ++-
.../storage/rocksdb/RocksDbMvPartitionStorage.java | 110 ++++++++++++++++-
.../storage/rocksdb/RocksDbTableStorage.java | 18 ++-
.../ignite/internal/storage/rocksdb/UuidRowId.java | 4 +-
.../distributed/ItInternalTableScanTest.java | 4 +-
.../ignite/distributed/ItTablePersistenceTest.java | 8 +-
.../distributed/ItTxDistributedTestSingleNode.java | 6 +-
.../ignite/internal/table/ItColocationTest.java | 4 +-
.../ignite/internal/table/InternalTable.java | 13 +-
.../internal/table/distributed/TableManager.java | 137 +++++++++++++--------
.../table/distributed/raft/PartitionListener.java | 18 ++-
.../raft/snapshot/InitPartitionSnapshotReader.java | 94 ++++++++++++++
.../raft/snapshot/PartitionSnapshotStorage.java | 123 ++++++++++++++++++
.../snapshot/PartitionSnapshotStorageFactory.java | 85 +++++++++++++
.../raft/snapshot/PartitionSnapshotWriter.java | 105 ++++++++++++++++
.../distributed/storage/InternalTableImpl.java | 15 ++-
.../distributed/storage/VersionedRowStore.java | 28 ++++-
.../org/apache/ignite/internal/table/Example.java | 3 +-
.../internal/table/InteropOperationsTest.java | 12 +-
.../table/KeyValueBinaryViewOperationsTest.java | 9 +-
.../KeyValueViewOperationsSimpleSchemaTest.java | 9 +-
.../internal/table/KeyValueViewOperationsTest.java | 10 +-
.../internal/table/MessagingServiceTestUtils.java | 21 +++-
.../table/RecordBinaryViewOperationsTest.java | 9 +-
.../internal/table/RecordViewOperationsTest.java | 10 +-
.../internal/table/SchemaValidationTest.java | 10 +-
.../apache/ignite/internal/table/TxLocalTest.java | 16 ++-
.../table/distributed/TableManagerTest.java | 6 +-
.../raft/PartitionCommandListenerTest.java | 30 +++++
.../table/impl/DummyInternalTableImpl.java | 19 ++-
49 files changed, 1094 insertions(+), 235 deletions(-)
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
index 7da5e0bc1..01dadc9d5 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
@@ -28,13 +28,12 @@ import java.util.concurrent.Flow.Publisher;
import javax.naming.OperationNotSupportedException;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.raft.client.service.RaftGroupService;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -63,7 +62,7 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull TableStorage storage() {
+ public MvTableStorage storage() {
throw new UnsupportedOperationException("Not implemented yet");
}
@@ -75,13 +74,13 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull UUID tableId() {
+ public UUID tableId() {
return tableId;
}
/** {@inheritDoc} */
@Override
- public @NotNull String name() {
+ public String name() {
return tableName;
}
@@ -266,13 +265,13 @@ public class FakeInternalTable implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx) {
+ public Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx) {
throw new IgniteInternalException(new OperationNotSupportedException());
}
/** {@inheritDoc} */
@Override
- public @NotNull List<String> assignments() {
+ public List<String> assignments() {
throw new IgniteInternalException(new OperationNotSupportedException());
}
diff --git a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommandClosure.java b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommandClosure.java
index 4ff50a765..f5f77b9a8 100644
--- a/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommandClosure.java
+++ b/modules/raft-client/src/main/java/org/apache/ignite/raft/client/service/CommandClosure.java
@@ -28,6 +28,14 @@ import org.jetbrains.annotations.Nullable;
* @see RaftGroupListener
*/
public interface CommandClosure<R extends Command> {
+ /**
+ * Corresponding log index of the command. Present for write commands only.
+ * Returns {@code 0} for read commands.
+ */
+ default long index() {
+ return 0;
+ }
+
/**
* Returns command.
*/
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
index 1e26ea10c..68e739b81 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ItLozaTest.java
@@ -34,6 +34,7 @@ import java.nio.file.Path;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.network.ClusterNode;
@@ -66,8 +67,16 @@ public class ItLozaTest {
* @return Raft group service.
*/
private RaftGroupService startClient(String groupId, ClusterNode node, Loza loza) throws Exception {
+ Supplier<RaftGroupListener> raftGroupListenerSupplier = () -> {
+ RaftGroupListener raftGroupListener = mock(RaftGroupListener.class);
+
+ when(raftGroupListener.onSnapshotLoad(any())).thenReturn(true);
+
+ return raftGroupListener;
+ };
+
return loza.prepareRaftGroup(groupId,
- List.of(node), () -> mock(RaftGroupListener.class), defaults()
+ List.of(node), raftGroupListenerSupplier, defaults()
).get(10, TimeUnit.SECONDS);
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
index bd60b3638..eb5f0c325 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
@@ -147,6 +147,15 @@ public class Loza implements IgniteComponent {
raftServer.stop();
}
+ /**
+ * Determines whether a RAFT group should be started locally according to a collection of nodes that should have a RAFT group.
+ */
+ public boolean shouldHaveRaftGroupLocally(Collection<ClusterNode> raftNodes) {
+ String locNodeName = clusterNetSvc.topologyService().localMember().name();
+
+ return raftNodes.stream().anyMatch(n -> locNodeName.equals(n.name()));
+ }
+
/**
* Creates a raft group service providing operations on a raft group. If {@code nodes} contains the current node, then raft group starts
* on the current node.
@@ -196,17 +205,14 @@ public class Loza implements IgniteComponent {
List<Peer> peers = nodes.stream().map(n -> new Peer(n.address())).collect(Collectors.toList());
- String locNodeName = clusterNetSvc.topologyService().localMember().name();
-
- boolean hasLocalRaft = nodes.stream().anyMatch(n -> locNodeName.equals(n.name()));
+ boolean hasLocalRaft = shouldHaveRaftGroupLocally(nodes);
if (hasLocalRaft) {
LOG.info("Start new raft node for group={} with initial peers={}", groupId, peers);
if (!raftServer.startRaftGroup(groupId, raftGrpEvtsLsnrSupplier.get(), lsnrSupplier.get(), peers, groupOptions)) {
throw new IgniteInternalException(IgniteStringFormatter.format(
- "Raft group on the node is already started [node={}, raftGrp={}]",
- locNodeName,
+ "Raft group on the node is already started [raftGrp={}]",
groupId
));
}
@@ -226,22 +232,20 @@ public class Loza implements IgniteComponent {
}
/**
- * If {@code deltaNodes} contains the current node, then raft group starts on the current node.
+ * Start RAFT group on the current node.
*
- * @param grpId Raft group id.
- * @param nodes Full set of raft group nodes.
- * @param deltaNodes New raft group nodes.
- * @param lsnrSupplier Raft group listener supplier.
- * @param raftGrpEvtsLsnrSupplier Raft group events listener supplier.
+ * @param grpId Raft group id.
+ * @param nodes Full set of raft group nodes.
+ * @param lsnr Raft group listener.
+ * @param raftGrpEvtsLsnr Raft group events listener.
* @param groupOptions Options to apply to the group.
* @throws NodeStoppingException If node stopping intention was detected.
*/
public void startRaftGroupNode(
String grpId,
Collection<ClusterNode> nodes,
- Collection<ClusterNode> deltaNodes,
- Supplier<RaftGroupListener> lsnrSupplier,
- Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier,
+ RaftGroupListener lsnr,
+ RaftGroupEventsListener raftGrpEvtsLsnr,
RaftGroupOptions groupOptions
) throws NodeStoppingException {
assert !nodes.isEmpty();
@@ -253,18 +257,13 @@ public class Loza implements IgniteComponent {
try {
List<Peer> peers = nodes.stream().map(n -> new Peer(n.address())).collect(Collectors.toList());
- String locNodeName = clusterNetSvc.topologyService().localMember().name();
-
- if (deltaNodes.stream().anyMatch(n -> locNodeName.equals(n.name()))) {
- LOG.info("Start new raft node for group={} with initial peers={}", grpId, peers);
+ LOG.info("Start new raft node for group={} with initial peers={}", grpId, peers);
- if (!raftServer.startRaftGroup(grpId, raftGrpEvtsLsnrSupplier.get(), lsnrSupplier.get(), peers, groupOptions)) {
- throw new IgniteInternalException(IgniteStringFormatter.format(
- "Raft group on the node is already started [node={}, raftGrp={}]",
- locNodeName,
- grpId
- ));
- }
+ if (!raftServer.startRaftGroup(grpId, raftGrpEvtsLsnr, lsnr, peers, groupOptions)) {
+ throw new IgniteInternalException(IgniteStringFormatter.format(
+ "Raft group on the node is already started [raftGrp={}]",
+ grpId
+ ));
}
} finally {
busyLock.leaveBusy();
@@ -272,87 +271,40 @@ public class Loza implements IgniteComponent {
}
/**
- * Creates a raft group service providing operations on a raft group. If {@code deltaNodes} contains the current node, then raft group
- * starts on the current node.
+ * Creates and starts a raft group service providing operations on a raft group.
*
- * @param grpId Raft group id.
- * @param nodes Full set of raft group nodes.
- * @param deltaNodes New raft group nodes.
- * @param lsnrSupplier Raft group listener supplier.
- * @param raftGrpEvtsLsnrSupplier Raft group events listener supplier.
- * @param groupOptions Options to apply to the group.
- * @return Future representing pending completion of the operation.
+ * @param grpId RAFT group id.
+ * @param nodes Full set of raft group nodes.
+ * @return Future that will be completed with an instance of RAFT group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
- public CompletableFuture<RaftGroupService> updateRaftGroup(
+ public CompletableFuture<RaftGroupService> startRaftGroupService(
String grpId,
- Collection<ClusterNode> nodes,
- Collection<ClusterNode> deltaNodes,
- Supplier<RaftGroupListener> lsnrSupplier,
- Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier,
- RaftGroupOptions groupOptions
+ Collection<ClusterNode> nodes
) throws NodeStoppingException {
if (!busyLock.enterBusy()) {
throw new NodeStoppingException();
}
+ List<Peer> peers = nodes.stream().map(n -> new Peer(n.address())).collect(Collectors.toList());
+
try {
- return updateRaftGroupInternal(grpId, nodes, deltaNodes, lsnrSupplier, raftGrpEvtsLsnrSupplier, groupOptions);
+ return RaftGroupServiceImpl.start(
+ grpId,
+ clusterNetSvc,
+ FACTORY,
+ RETRY_TIMEOUT,
+ RPC_TIMEOUT,
+ peers,
+ true,
+ DELAY,
+ executor
+ );
} finally {
busyLock.leaveBusy();
}
}
- /**
- * Internal method for updating a raft group.
- *
- * @param grpId Raft group id.
- * @param nodes Full set of raft group nodes.
- * @param deltaNodes New raft group nodes.
- * @param lsnrSupplier Raft group listener supplier.
- * @param raftGrpEvtsLsnrSupplier Raft group events listener supplier.
- * @param groupOptions Options to apply to the group.
- * @return Future representing pending completion of the operation.
- */
- private CompletableFuture<RaftGroupService> updateRaftGroupInternal(
- String grpId,
- Collection<ClusterNode> nodes,
- Collection<ClusterNode> deltaNodes,
- Supplier<RaftGroupListener> lsnrSupplier,
- Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier,
- RaftGroupOptions groupOptions
- ) {
- assert !nodes.isEmpty();
-
- List<Peer> peers = nodes.stream().map(n -> new Peer(n.address())).collect(Collectors.toList());
-
- String locNodeName = clusterNetSvc.topologyService().localMember().name();
-
- if (deltaNodes.stream().anyMatch(n -> locNodeName.equals(n.name()))) {
- LOG.info("Start new raft node for group={} with initial peers={}", grpId, peers);
-
- if (!raftServer.startRaftGroup(grpId, raftGrpEvtsLsnrSupplier.get(), lsnrSupplier.get(), peers, groupOptions)) {
- throw new IgniteInternalException(IgniteStringFormatter.format(
- "Raft group on the node is already started [node={}, raftGrp={}]",
- locNodeName,
- grpId
- ));
- }
- }
-
- return RaftGroupServiceImpl.start(
- grpId,
- clusterNetSvc,
- FACTORY,
- RETRY_TIMEOUT,
- RPC_TIMEOUT,
- peers,
- true,
- DELAY,
- executor
- );
- }
-
/**
* Stops a raft group on the current node.
*
diff --git a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
index 6081e65a5..8b36e8e31 100644
--- a/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/internal/raft/server/impl/JraftServerImpl.java
@@ -499,14 +499,25 @@ public class JraftServerImpl implements RaftServer {
public CommandClosure<WriteCommand> next() {
@Nullable CommandClosure<WriteCommand> done = (CommandClosure<WriteCommand>) iter.done();
ByteBuffer data = iter.getData();
- WriteCommand command = JDKMarshaller.DEFAULT.unmarshall(data.array());
+
+ WriteCommand command = done == null ? JDKMarshaller.DEFAULT.unmarshall(data.array()) : done.command();
+
+ long commandIndex = iter.getIndex();
return new CommandClosure<>() {
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
@Override
public WriteCommand command() {
return command;
}
+ /** {@inheritDoc} */
@Override
public void result(Serializable res) {
if (done != null) {
@@ -543,9 +554,14 @@ public class JraftServerImpl implements RaftServer {
if (res == null) {
File file = new File(writer.getPath());
- for (File file0 : file.listFiles()) {
- if (file0.isFile()) {
- writer.addFile(file0.getName(), null);
+ File[] snapshotFiles = file.listFiles();
+
+ // Files array can be null if shanpshot folder doesn't exist.
+ if (snapshotFiles != null) {
+ for (File file0 : snapshotFiles) {
+ if (file0.isFile()) {
+ writer.addFile(file0.getName(), null);
+ }
}
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/snapshot/SnapshotExecutorImpl.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/snapshot/SnapshotExecutorImpl.java
index c5939f083..00f27c56b 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/snapshot/SnapshotExecutorImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/storage/snapshot/SnapshotExecutorImpl.java
@@ -228,9 +228,11 @@ public class SnapshotExecutorImpl implements SnapshotExecutor {
LOG.error("Fail to init snapshot storage.");
return false;
}
- final LocalSnapshotStorage tmp = (LocalSnapshotStorage) this.snapshotStorage;
- if (tmp != null && !tmp.hasServerAddr()) {
- tmp.setServerAddr(opts.getAddr());
+ if (snapshotStorage instanceof LocalSnapshotStorage) {
+ final LocalSnapshotStorage tmp = (LocalSnapshotStorage) this.snapshotStorage;
+ if (tmp != null && !tmp.hasServerAddr()) {
+ tmp.setServerAddr(opts.getAddr());
+ }
}
final SnapshotReader reader = this.snapshotStorage.open();
if (reader == null) {
diff --git a/modules/raft/src/test/java/org/apache/ignite/internal/raft/LozaTest.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/LozaTest.java
index b13450a3c..91b1c03f7 100644
--- a/modules/raft/src/test/java/org/apache/ignite/internal/raft/LozaTest.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/LozaTest.java
@@ -80,7 +80,7 @@ public class LozaTest extends IgniteAbstractTest {
assertThrows(
NodeStoppingException.class,
- () -> loza.updateRaftGroup(raftGroupId, nodes, newNodes, lsnrSupplier, () -> null, defaults())
+ () -> loza.startRaftGroupService(raftGroupId, newNodes)
);
assertThrows(NodeStoppingException.class, () -> loza.stopRaftGroup(raftGroupId));
assertThrows(
diff --git a/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java b/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
index 02149c65b..659d75730 100644
--- a/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
+++ b/modules/raft/src/test/java/org/apache/ignite/internal/raft/server/impl/RaftServerImpl.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -72,6 +73,8 @@ public class RaftServerImpl implements RaftServer {
private final BlockingQueue<CommandClosureEx<WriteCommand>> writeQueue;
+ private final AtomicLong raftIndex = new AtomicLong();
+
private volatile Thread readWorker;
private volatile Thread writeWorker;
@@ -232,17 +235,28 @@ public class RaftServerImpl implements RaftServer {
BlockingQueue<CommandClosureEx<T>> queue,
RaftGroupListener lsnr
) {
+ long commandIndex = req.command() instanceof ReadCommand ? 0 : raftIndex.incrementAndGet();
+
if (!queue.offer(new CommandClosureEx<>() {
+ /** {@inheritDoc} */
@Override
public RaftGroupListener listener() {
return lsnr;
}
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
@Override
public T command() {
return (T) req.command();
}
+ /** {@inheritDoc} */
@Override
public void result(Serializable res) {
NetworkMessage msg;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
index 18f3b0f26..cd9258cad 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/schema/IgniteTableImpl.java
@@ -61,7 +61,7 @@ import org.apache.ignite.internal.sql.engine.trait.RewindabilityTrait;
import org.apache.ignite.internal.sql.engine.type.IgniteTypeFactory;
import org.apache.ignite.internal.sql.engine.util.Commons;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
-import org.apache.ignite.internal.storage.PartitionStorage;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.jetbrains.annotations.Nullable;
@@ -477,7 +477,7 @@ public class IgniteTableImpl extends AbstractTable implements InternalIgniteTabl
long size = 0L;
for (int p = 0; p < parts; ++p) {
- @Nullable PartitionStorage part = table.storage().getPartition(p);
+ @Nullable MvPartitionStorage part = table.storage().getMvPartition(p);
if (part != null) {
size += part.rowsCount();
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
index 1df3614cf..4998f6199 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
@@ -67,7 +67,7 @@ import org.apache.ignite.internal.schema.registry.SchemaRegistryImpl;
import org.apache.ignite.internal.schema.row.RowAssembler;
import org.apache.ignite.internal.sql.engine.exec.ExecutionCancelledException;
import org.apache.ignite.internal.storage.DataStorageManager;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.distributed.TableManager;
@@ -220,7 +220,7 @@ public class StopCalciteModuleTest {
when(tbl.tableId()).thenReturn(UUID.randomUUID());
- when(tbl.storage()).thenReturn(mock(TableStorage.class));
+ when(tbl.storage()).thenReturn(mock(MvTableStorage.class));
when(tbl.storage().configuration()).thenReturn(mock(TableConfiguration.class));
when(tbl.storage().configuration().partitions()).thenReturn(mock(ConfigurationValue.class));
when(tbl.storage().configuration().partitions().value()).thenReturn(1);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
index fb26c9b5d..05cfc52c7 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/MockedStructuresTest.java
@@ -650,7 +650,7 @@ public class MockedStructuresTest extends IgniteAbstractTest {
return completedFuture(raftGrpSrvcMock);
});
- when(rm.updateRaftGroup(any(), any(), any(), any(), any(), any())).thenAnswer(mock -> {
+ when(rm.startRaftGroupService(any(), any())).thenAnswer(mock -> {
RaftGroupService raftGrpSrvcMock = mock(RaftGroupService.class);
when(raftGrpSrvcMock.leader()).thenReturn(new Peer(new NetworkAddress("localhost", 47500)));
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 31c9c578f..7eb39e8f0 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
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.storage;
import java.util.UUID;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.tx.Timestamp;
@@ -30,6 +31,21 @@ import org.jetbrains.annotations.Nullable;
* <p>Each MvPartitionStorage instance represents exactly one partition.
*/
public interface MvPartitionStorage extends AutoCloseable {
+ /**
+ * Index of the highest write command applied to the storage. {@code 0} if index is unknown.
+ */
+ long lastAppliedIndex();
+
+ /**
+ * Sets the last applied index value.
+ */
+ void lastAppliedIndex(long lastAppliedIndex) throws StorageException;
+
+ /**
+ * {@link #lastAppliedIndex()} value consistent with the data, already persisted on the storage.
+ */
+ long persistedIndex();
+
/**
* Reads either the committed value from the storage or the uncommitted value belonging to given transaction.
*
@@ -118,4 +134,25 @@ public interface MvPartitionStorage extends AutoCloseable {
* @throws StorageException If failed to read data from the storage.
*/
Cursor<BinaryRow> scan(Predicate<BinaryRow> keyFilter, Timestamp timestamp) throws StorageException;
+
+ /**
+ * Returns rows count belongs to current storage.
+ *
+ * @return Rows count.
+ * @throws StorageException If failed to obtain size.
+ * @deprecated It's not yet defined what a "count" is. This value is not easily defined for multiversioned storages.
+ * TODO IGNITE-16769 Implement correct PartitionStorage rows count calculation.
+ */
+ @Deprecated
+ long rowsCount() throws StorageException;
+
+ /**
+ * Iterates over all versions of all entries, except for tombstones.
+ *
+ * @param consumer Closure to process entries.
+ * @deprecated This method was bord out of desperation and isn't well-designed. Implementation is not polished either. Currently, it's
+ * only usage is to work-around in-memory PK index rebuild on node restart, which shouldn't even exist in the first place.
+ */
+ @Deprecated
+ void forEach(BiConsumer<RowId, BinaryRow> consumer);
}
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/PartitionStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/PartitionStorage.java
index 2a7d6f27a..53e362bd3 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/PartitionStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/PartitionStorage.java
@@ -28,7 +28,10 @@ import org.jetbrains.annotations.Nullable;
/**
* Interface providing methods to read, remove and update keys in storage. Any locking is unnecessary as this storage is used within RAFT
* groups where all write operations are serialized.
+ *
+ * @deprecated Replaced with {@link MvPartitionStorage}.
*/
+@Deprecated
public interface PartitionStorage extends AutoCloseable {
/**
* Returns the partition id.
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/MvTableStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/MvTableStorage.java
index f92e0684e..7cfcbde45 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/MvTableStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/MvTableStorage.java
@@ -55,6 +55,11 @@ public interface MvTableStorage {
*/
CompletableFuture<?> destroyPartition(int partitionId) throws StorageException;
+ /**
+ * Returns {@code true} if this storage is volatile (i.e. stores its data in memory), or {@code false} if it's persistent.
+ */
+ boolean isVolatile();
+
/**
* Returns the table configuration.
*/
diff --git a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/TableStorage.java b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/TableStorage.java
index 7ab3ec975..29e76b12f 100644
--- a/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/TableStorage.java
+++ b/modules/storage-api/src/main/java/org/apache/ignite/internal/storage/engine/TableStorage.java
@@ -25,7 +25,10 @@ import org.jetbrains.annotations.Nullable;
/**
* Table storage that contains meta, partitions and SQL indexes.
+ *
+ * @deprecated Replaced with {@link MvTableStorage}.
*/
+@Deprecated
public interface TableStorage {
/**
* Retrieves or creates a partition for the current table. Not expected to be called concurrently with the same partition id.
diff --git a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/basic/TestMvPartitionStorage.java b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/basic/TestMvPartitionStorage.java
index 4670cf031..a43190b25 100644
--- a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/basic/TestMvPartitionStorage.java
+++ b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/basic/TestMvPartitionStorage.java
@@ -19,10 +19,12 @@ package org.apache.ignite.internal.storage.basic;
import java.util.Iterator;
import java.util.List;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.storage.MvPartitionStorage;
@@ -41,6 +43,8 @@ public class TestMvPartitionStorage implements MvPartitionStorage {
private final List<TestSortedIndexMvStorage> indexes;
+ private long lastAppliedIndex = 0;
+
private final int partitionId;
public TestMvPartitionStorage(List<TestSortedIndexMvStorage> indexes, int partitionId) {
@@ -74,6 +78,26 @@ public class TestMvPartitionStorage implements MvPartitionStorage {
}
}
+ /** {@inheritDoc} */
+ @Override
+ public long lastAppliedIndex() {
+ return lastAppliedIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void lastAppliedIndex(long lastAppliedIndex) throws StorageException {
+ assert lastAppliedIndex > this.lastAppliedIndex : "current=" + this.lastAppliedIndex + ", new=" + lastAppliedIndex;
+
+ this.lastAppliedIndex = lastAppliedIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long persistedIndex() {
+ return lastAppliedIndex;
+ }
+
/** {@inheritDoc} */
@Override
public RowId insert(BinaryRow row, UUID txId) throws StorageException {
@@ -251,6 +275,30 @@ public class TestMvPartitionStorage implements MvPartitionStorage {
return Cursor.fromIterator(iterator);
}
+ /** {@inheritDoc} */
+ @Override
+ public long rowsCount() {
+ return map.size();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void forEach(BiConsumer<RowId, BinaryRow> consumer) {
+ for (Entry<RowId, VersionChain> entry : map.entrySet()) {
+ RowId rowId = entry.getKey();
+
+ VersionChain versionChain = entry.getValue();
+
+ for (VersionChain cur = versionChain; cur != null; cur = cur.next) {
+ if (cur.row == null) {
+ continue;
+ }
+
+ consumer.accept(rowId, cur.row);
+ }
+ }
+ }
+
/** {@inheritDoc} */
@Override
public void close() throws Exception {
diff --git a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/chm/TestMvTableStorage.java b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/chm/TestMvTableStorage.java
index 9df6a1d23..85746bc3a 100644
--- a/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/chm/TestMvTableStorage.java
+++ b/modules/storage-api/src/test/java/org/apache/ignite/internal/storage/chm/TestMvTableStorage.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.storage.chm;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.ignite.configuration.schemas.table.TableConfiguration;
@@ -55,7 +54,7 @@ public class TestMvTableStorage implements MvTableStorage {
@Override
public MvPartitionStorage getMvPartition(int partitionId) {
- return Objects.requireNonNull(partitions.get(partitionId), "Partition doesn't exist");
+ return partitions.get(partitionId);
}
@Override
@@ -65,6 +64,11 @@ public class TestMvTableStorage implements MvTableStorage {
return CompletableFuture.completedFuture(null);
}
+ @Override
+ public boolean isVolatile() {
+ return true;
+ }
+
@Override
public TableConfiguration configuration() {
return tableConfig;
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryStorageEngine.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryStorageEngine.java
index fa0b7c4c2..3af642d20 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryStorageEngine.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/VolatilePageMemoryStorageEngine.java
@@ -102,6 +102,12 @@ public class VolatilePageMemoryStorageEngine implements StorageEngine {
return new VolatilePageMemoryTableStorage(tableCfg, regions.get(dataStorageView.dataRegion()));
}
+ /** {@inheritDoc} */
+ @Override
+ public VolatilePageMemoryTableStorage createMvTable(TableConfiguration tableCfg) throws StorageException {
+ return createTable(tableCfg);
+ }
+
/**
* Creates, starts and adds a new data region to the engine.
*
diff --git a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
index d9eab0ec9..26ddc159f 100644
--- a/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
+++ b/modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PageMemoryMvPartitionStorage.java
@@ -24,6 +24,7 @@ import java.util.NoSuchElementException;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.configuration.schemas.table.TableView;
import org.apache.ignite.internal.pagememory.DataRegion;
@@ -70,6 +71,11 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
ScanVersionChainByTimestamp::new
);
+ /**
+ * Last applied index value.
+ */
+ private volatile long lastAppliedIndex = 0;
+
/**
* Constructor.
*/
@@ -122,6 +128,24 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
);
}
+ /** {@inheritDoc} */
+ @Override
+ public long lastAppliedIndex() {
+ return lastAppliedIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void lastAppliedIndex(long lastAppliedIndex) throws StorageException {
+ this.lastAppliedIndex = lastAppliedIndex;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long persistedIndex() {
+ return lastAppliedIndex;
+ }
+
/** {@inheritDoc} */
@Override
public @Nullable BinaryRow read(RowId rowId, UUID txId) throws TxIdMismatchException, StorageException {
@@ -466,6 +490,22 @@ public class PageMemoryMvPartitionStorage implements MvPartitionStorage {
return new ScanCursor(treeCursor, keyFilter, transactionId, timestamp);
}
+ /** {@inheritDoc} */
+ @Override
+ public long rowsCount() {
+ try {
+ return versionChainTree.size();
+ } catch (IgniteInternalCheckedException e) {
+ throw new StorageException("Error occurred while fetching the size.", e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void forEach(BiConsumer<RowId, BinaryRow> consumer) {
+ // No-op. Nothing to recover for a volatile storage. See usages and a comment about PK index rebuild.
+ }
+
/** {@inheritDoc} */
@Override
public void close() {
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
index cb01fea61..25c6b148a 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbMetaStorage.java
@@ -51,10 +51,17 @@ class RocksDbMetaStorage {
PARTITION_ID_PREFIX_END[PARTITION_ID_PREFIX_END.length - 1] += 1;
}
- private final ColumnFamily metaCf;
+ private final ColumnFamily metaColumnFamily;
- RocksDbMetaStorage(ColumnFamily metaCf) {
- this.metaCf = metaCf;
+ RocksDbMetaStorage(ColumnFamily metaColumnFamily) {
+ this.metaColumnFamily = metaColumnFamily;
+ }
+
+ /**
+ * Returns a column family instance, associated with the meta storage.
+ */
+ ColumnFamily columnFamily() {
+ return metaColumnFamily;
}
/**
@@ -68,7 +75,7 @@ class RocksDbMetaStorage {
try (
var upperBound = new Slice(PARTITION_ID_PREFIX_END);
var options = new ReadOptions().setIterateUpperBound(upperBound);
- RocksIterator it = metaCf.newIterator(options);
+ RocksIterator it = metaColumnFamily.newIterator(options);
) {
it.seek(PARTITION_ID_PREFIX);
@@ -102,7 +109,7 @@ class RocksDbMetaStorage {
byte[] partitionIdBytes = Arrays.copyOfRange(partitionIdKey, PARTITION_ID_PREFIX.length, partitionIdKey.length);
try {
- metaCf.put(partitionIdKey, partitionIdBytes);
+ metaColumnFamily.put(partitionIdKey, partitionIdBytes);
} catch (RocksDBException e) {
throw new StorageException("Unable to save partition " + partitionId + " in the meta Column Family", e);
}
@@ -115,7 +122,7 @@ class RocksDbMetaStorage {
*/
void removePartitionId(int partitionId) {
try {
- metaCf.delete(partitionIdKey(partitionId));
+ metaColumnFamily.delete(partitionIdKey(partitionId));
} catch (RocksDBException e) {
throw new StorageException("Unable to delete partition " + partitionId + " from the meta Column Family", e);
}
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 780c6bd1a..9a3235b4e 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
@@ -23,9 +23,11 @@ import static java.nio.ByteOrder.BIG_ENDIAN;
import static java.nio.ByteOrder.LITTLE_ENDIAN;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.NoSuchElementException;
import java.util.UUID;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.ByteBufferRow;
@@ -34,6 +36,7 @@ import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.StorageException;
import org.apache.ignite.internal.storage.TxIdMismatchException;
import org.apache.ignite.internal.tx.Timestamp;
+import org.apache.ignite.internal.util.ByteUtils;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.IgniteUtils;
@@ -95,23 +98,31 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
/** Partitions column family. */
private final ColumnFamilyHandle cf;
+ /** Meta column family. */
+ private final ColumnFamilyHandle meta;
+
/** Write options. */
private final WriteOptions writeOpts = new WriteOptions();
/** Upper bound for scans and reads. */
private final Slice upperBound;
+ /** Key to store applied index value in meta. */
+ private byte[] lastAppliedIndexKey;
+
/**
* Constructor.
*
* @param partitionId Partition id.
* @param db RocksDB instance.
* @param cf Column family handle to store partition data.
+ * @param meta Column family handle to store partition metadata.
*/
- public RocksDbMvPartitionStorage(int partitionId, RocksDB db, ColumnFamilyHandle cf) {
+ public RocksDbMvPartitionStorage(int partitionId, RocksDB db, ColumnFamilyHandle cf, ColumnFamilyHandle meta) {
this.partitionId = partitionId;
this.db = db;
this.cf = cf;
+ this.meta = meta;
heapKeyBuffer = withInitial(() ->
ByteBuffer.allocate(MAX_KEY_SIZE)
@@ -119,6 +130,42 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
);
upperBound = new Slice(partitionEndPrefix());
+
+ lastAppliedIndexKey = ("index" + partitionId).getBytes(StandardCharsets.UTF_8);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long lastAppliedIndex() {
+ byte[] appliedIndexBytes;
+
+ try {
+ appliedIndexBytes = db.get(meta, lastAppliedIndexKey);
+ } catch (RocksDBException e) {
+ throw new StorageException(e);
+ }
+
+ if (appliedIndexBytes == null) {
+ return 0;
+ }
+
+ return ByteUtils.bytesToLong(appliedIndexBytes);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void lastAppliedIndex(long lastAppliedIndex) throws StorageException {
+ try {
+ db.put(meta, lastAppliedIndexKey, ByteUtils.longToBytes(lastAppliedIndex));
+ } catch (RocksDBException e) {
+ throw new StorageException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public long persistedIndex() {
+ return lastAppliedIndex();
}
/** {@inheritDoc} */
@@ -528,6 +575,67 @@ public class RocksDbMvPartitionStorage implements MvPartitionStorage {
};
}
+ @Override
+ public long rowsCount() {
+ try (
+ var upperBound = new Slice(partitionEndPrefix());
+ var options = new ReadOptions().setIterateUpperBound(upperBound);
+ RocksIterator it = db.newIterator(cf, options)
+ ) {
+ it.seek(partitionStartPrefix());
+
+ long size = 0;
+
+ while (it.isValid()) {
+ ++size;
+ it.next();
+ }
+
+ return size;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void forEach(BiConsumer<RowId, BinaryRow> consumer) {
+ try (
+ var upperBound = new Slice(partitionEndPrefix());
+ var options = new ReadOptions().setIterateUpperBound(upperBound);
+ RocksIterator it = db.newIterator(cf, options)
+ ) {
+ it.seek(partitionStartPrefix());
+
+ while (it.isValid()) {
+ byte[] keyBytes = it.key();
+ byte[] valueBytes = it.value();
+
+ boolean valueHasTxId = keyBytes.length == ROW_PREFIX_SIZE;
+
+ if (!isTombstone(valueBytes, valueHasTxId)) {
+ ByteBuffer keyBuf = ByteBuffer.wrap(keyBytes).order(BIG_ENDIAN);
+ RowId rowId = new UuidRowId(keyBuf.getLong(), keyBuf.getLong());
+
+ BinaryRow binaryRow = wrapValueIntoBinaryRow(valueBytes, valueHasTxId);
+
+ consumer.accept(rowId, binaryRow);
+ }
+
+ it.next();
+ }
+ }
+ }
+
+ /**
+ * Deletes partition data from the storage.
+ */
+ public void destroy() {
+ try {
+ db.delete(meta, lastAppliedIndexKey);
+ } catch (RocksDBException e) {
+ throw new StorageException("Failed to cleanup partition meta on destruction", e);
+ }
+ }
+
/** {@inheritDoc} */
@Override
public void close() throws Exception {
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
index 8a94317ca..02b35530a 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/RocksDbTableStorage.java
@@ -258,6 +258,7 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
}
}
+ /** {@inheritDoc} */
@Override
public RocksDbMvPartitionStorage getOrCreateMvPartition(int partitionId) throws StorageException {
getOrCreatePartition(partitionId);
@@ -265,22 +266,31 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
return getMvPartition(partitionId);
}
+ /** {@inheritDoc} */
@Override
- public RocksDbMvPartitionStorage getMvPartition(int partitionId) {
+ public @Nullable RocksDbMvPartitionStorage getMvPartition(int partitionId) {
if (getPartition(partitionId) == null) {
- throw new NullPointerException("Partition doesn't exist");
+ return null;
}
- return new RocksDbMvPartitionStorage(partitionId, db, partitionCf.handle());
+ return new RocksDbMvPartitionStorage(partitionId, db, partitionCf.handle(), meta.columnFamily().handle());
}
+ /** {@inheritDoc} */
@Override
public CompletableFuture<?> destroyPartition(int partitionId) throws StorageException {
+ RocksDbMvPartitionStorage mvPartition = getMvPartition(partitionId);
+
dropPartition(partitionId);
+ if (mvPartition != null) {
+ mvPartition.destroy();
+ }
+
return CompletableFuture.completedFuture(null);
}
+ /** {@inheritDoc} */
@Override
public SortedIndexStorage getOrCreateSortedIndex(String indexName) {
assert !stopped : "Storage has been stopped";
@@ -301,6 +311,7 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
});
}
+ /** {@inheritDoc} */
@Override
public void dropIndex(String indexName) {
assert !stopped : "Storage has been stopped";
@@ -312,6 +323,7 @@ class RocksDbTableStorage implements TableStorage, MvTableStorage {
});
}
+ /** {@inheritDoc} */
@Override
public boolean isVolatile() {
return false;
diff --git a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/UuidRowId.java b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/UuidRowId.java
index a41779c06..c252a159a 100644
--- a/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/UuidRowId.java
+++ b/modules/storage-rocksdb/src/main/java/org/apache/ignite/internal/storage/rocksdb/UuidRowId.java
@@ -25,6 +25,8 @@ import org.apache.ignite.internal.storage.RowId;
/**
* UUID-based ignite row id implementation.
*/
+@Deprecated
+//TODO Implement generic row id. IGNITE-17076
final class UuidRowId implements RowId {
/*
* The most significant 64 bits.
@@ -37,7 +39,7 @@ final class UuidRowId implements RowId {
private final long leastSigBits;
/** Private constructor. */
- private UuidRowId(long mostSigBits, long leastSigBits) {
+ public UuidRowId(long mostSigBits, long leastSigBits) {
this.mostSigBits = mostSigBits;
this.leastSigBits = leastSigBits;
}
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
index 36487a2d0..59b93e9a0 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItInternalTableScanTest.java
@@ -57,7 +57,7 @@ import org.apache.ignite.internal.schema.ByteBufferRow;
import org.apache.ignite.internal.storage.DataRow;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.StorageException;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
@@ -190,7 +190,7 @@ public class ItInternalTableScanTest {
NetworkAddress::toString,
addressToNode,
txManager,
- mock(TableStorage.class)
+ mock(MvTableStorage.class)
);
}
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 b2e68f753..749f52abc 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
@@ -39,7 +39,7 @@ import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.row.Row;
import org.apache.ignite.internal.schema.row.RowAssembler;
import org.apache.ignite.internal.storage.basic.TestMvPartitionStorage;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.distributed.TableTxManagerImpl;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
@@ -113,7 +113,7 @@ public class ItTablePersistenceTest extends ItAbstractListenerSnapshotTest<Parti
NetworkAddress::toString,
addressToNode,
txManager,
- mock(TableStorage.class)
+ mock(MvTableStorage.class)
);
table.upsert(FIRST_VALUE, null).get();
@@ -136,7 +136,7 @@ public class ItTablePersistenceTest extends ItAbstractListenerSnapshotTest<Parti
NetworkAddress::toString,
addressToNode,
txManager,
- mock(TableStorage.class)
+ mock(MvTableStorage.class)
);
// Remove the first key
@@ -165,7 +165,7 @@ public class ItTablePersistenceTest extends ItAbstractListenerSnapshotTest<Parti
NetworkAddress::toString,
addressToNode,
txManager,
- mock(TableStorage.class)
+ mock(MvTableStorage.class)
);
table.upsert(SECOND_VALUE, null).get();
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 6dff9cf15..2f4c64ea4 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
@@ -41,7 +41,7 @@ import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.raft.server.RaftGroupOptions;
import org.apache.ignite.internal.raft.server.impl.JraftServerImpl;
import org.apache.ignite.internal.storage.basic.TestMvPartitionStorage;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.TxAbstractTest;
import org.apache.ignite.internal.table.distributed.TableTxManagerImpl;
@@ -229,7 +229,7 @@ public class ItTxDistributedTestSingleNode extends TxAbstractTest {
NetworkAddress::toString,
addressToNode,
txMgr,
- Mockito.mock(TableStorage.class)
+ Mockito.mock(MvTableStorage.class)
), new DummySchemaManagerImpl(ACCOUNTS_SCHEMA));
this.customers = new TableImpl(new InternalTableImpl(
@@ -240,7 +240,7 @@ public class ItTxDistributedTestSingleNode extends TxAbstractTest {
NetworkAddress::toString,
addressToNode,
txMgr,
- Mockito.mock(TableStorage.class)
+ Mockito.mock(MvTableStorage.class)
), new DummySchemaManagerImpl(CUSTOMERS_SCHEMA));
log.info("Tables have been started");
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
index 038493744..4147bcc64 100644
--- a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
@@ -57,7 +57,7 @@ import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerException;
import org.apache.ignite.internal.schema.marshaller.TupleMarshallerImpl;
import org.apache.ignite.internal.schema.row.Row;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.distributed.command.InsertCommand;
import org.apache.ignite.internal.table.distributed.command.MultiKeyCommand;
import org.apache.ignite.internal.table.distributed.command.response.MultiRowsResponse;
@@ -150,7 +150,7 @@ public class ItColocationTest {
null,
null,
txManager,
- Mockito.mock(TableStorage.class)
+ Mockito.mock(MvTableStorage.class)
);
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
index 10cc05936..f05cfe846 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
@@ -24,12 +24,11 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow.Publisher;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.LockException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.raft.client.service.RaftGroupService;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
/**
@@ -42,21 +41,21 @@ public interface InternalTable extends AutoCloseable {
*
* @return Table storage.
*/
- @NotNull TableStorage storage();
+ MvTableStorage storage();
/**
* Gets a table id.
*
* @return Table id as UUID.
*/
- @NotNull UUID tableId();
+ UUID tableId();
/**
* Gets a name of the table.
*
* @return Table name.
*/
- @NotNull String name();
+ String name();
/**
* Asynchronously gets a row with same key columns values as given one from the table.
@@ -211,7 +210,7 @@ public interface InternalTable extends AutoCloseable {
* @param tx The transaction.
* @return {@link Publisher} that reactively notifies about partition rows.
*/
- @NotNull Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx);
+ Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx);
/**
* Gets a count of partitions of the table.
@@ -228,7 +227,7 @@ public interface InternalTable extends AutoCloseable {
*
* @return List of current assignments.
*/
- @NotNull List<String> assignments();
+ List<String> assignments();
/**
* Returns cluster node that is the leader of the corresponding partition group or throws an exception if
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 4a9b5abed..e0520a290 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
@@ -54,7 +54,6 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
import org.apache.ignite.configuration.ConfigurationChangeException;
import org.apache.ignite.configuration.ConfigurationProperty;
import org.apache.ignite.configuration.notifications.ConfigurationNamedListListener;
@@ -91,13 +90,14 @@ import org.apache.ignite.internal.schema.event.SchemaEvent;
import org.apache.ignite.internal.schema.event.SchemaEventParameters;
import org.apache.ignite.internal.schema.marshaller.schema.SchemaSerializerImpl;
import org.apache.ignite.internal.storage.DataStorageManager;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.engine.MvTableStorage;
-import org.apache.ignite.internal.storage.engine.TableStorage;
import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import org.apache.ignite.internal.table.distributed.raft.RebalanceRaftGroupEventsListener;
+import org.apache.ignite.internal.table.distributed.raft.snapshot.PartitionSnapshotStorageFactory;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
import org.apache.ignite.internal.table.distributed.storage.VersionedRowStore;
import org.apache.ignite.internal.table.event.TableEvent;
@@ -122,6 +122,7 @@ import org.apache.ignite.network.TopologyService;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupListener;
import org.apache.ignite.raft.client.service.RaftGroupService;
+import org.apache.ignite.raft.jraft.entity.PeerId;
import org.apache.ignite.raft.jraft.storage.impl.VolatileRaftMetaStorage;
import org.apache.ignite.raft.jraft.util.Utils;
import org.apache.ignite.raft.jraft.util.concurrent.ConcurrentHashSet;
@@ -484,24 +485,37 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
assert internalTbl.storage() instanceof MvTableStorage :
"Only multi version storages are supported. Current storage is a " + internalTbl.storage().getClass().getName();
- futures[partId] = raftMgr.updateRaftGroup(
- partitionRaftGroupName(tblId, partId),
- newPartAssignment,
- // start new nodes, only if it is table creation
- // other cases will be covered by rebalance logic
- (oldPartAssignment.isEmpty()) ? newPartAssignment : Collections.emptyList(),
- () -> new PartitionListener(tblId,
- new VersionedRowStore(((MvTableStorage) internalTbl.storage()).getOrCreateMvPartition(partId),
- txManager)),
- () -> new RebalanceRaftGroupEventsListener(
- metaStorageMgr,
- tablesCfg.tables().get(tablesById.get(tblId).name()),
- partitionRaftGroupName(tblId, partId),
- partId,
- busyLock,
- movePartition(() -> internalTbl.partitionRaftGroupService(partId)),
- rebalanceScheduler),
- groupOptionsForInternalTable(internalTbl)
+ // start new nodes, only if it is table creation
+ // other cases will be covered by rebalance logic
+ List<ClusterNode> nodes = (oldPartAssignment.isEmpty()) ? newPartAssignment : Collections.emptyList();
+
+ String grpId = partitionRaftGroupName(tblId, partId);
+
+ if (raftMgr.shouldHaveRaftGroupLocally(nodes)) {
+ MvPartitionStorage partitionStorage = internalTbl.storage().getOrCreateMvPartition(partId);
+
+ RaftGroupOptions groupOptions = groupOptionsForPartition(internalTbl, partitionStorage, newPartAssignment);
+
+ raftMgr.startRaftGroupNode(
+ grpId,
+ newPartAssignment,
+ new PartitionListener(tblId, new VersionedRowStore(partitionStorage, txManager)),
+ new RebalanceRaftGroupEventsListener(
+ metaStorageMgr,
+ tablesCfg.tables().get(tablesById.get(tblId).name()),
+ grpId,
+ partId,
+ busyLock,
+ movePartition(() -> internalTbl.partitionRaftGroupService(partId)),
+ rebalanceScheduler
+ ),
+ groupOptions
+ );
+ }
+
+ futures[partId] = raftMgr.startRaftGroupService(
+ grpId,
+ newPartAssignment
).thenAccept(
updatedRaftGroupService -> ((InternalTableImpl) internalTbl)
.updateInternalTableRaftGroupService(partId, updatedRaftGroupService)
@@ -521,14 +535,29 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
CompletableFuture.allOf(futures).join();
}
- private RaftGroupOptions groupOptionsForInternalTable(InternalTable internalTbl) {
+ private RaftGroupOptions groupOptionsForPartition(
+ InternalTable internalTbl,
+ MvPartitionStorage partitionStorage,
+ List<ClusterNode> peers
+ ) {
+ RaftGroupOptions raftGroupOptions;
+
if (internalTbl.storage().isVolatile()) {
- return RaftGroupOptions.forVolatileStores()
+ raftGroupOptions = RaftGroupOptions.forVolatileStores()
.setLogStorageFactory(new VolatileLogStorageFactory())
.raftMetaStorageFactory((groupId, raftOptions) -> new VolatileRaftMetaStorage());
} else {
- return RaftGroupOptions.forPersistentStores();
+ raftGroupOptions = RaftGroupOptions.forPersistentStores();
}
+
+ //TODO Revisit peers String representation: https://issues.apache.org/jira/browse/IGNITE-17420
+ raftGroupOptions.snapshotStorageFactory(new PartitionSnapshotStorageFactory(
+ partitionStorage,
+ peers.stream().map(n -> new Peer(n.address())).map(PeerId::fromPeer).map(Object::toString).collect(Collectors.toList()),
+ List.of()
+ ));
+
+ return raftGroupOptions;
}
/** {@inheritDoc} */
@@ -544,12 +573,12 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
for (TableImpl table : tables.values()) {
try {
- table.internalTable().storage().stop();
- table.internalTable().close();
-
for (int p = 0; p < table.internalTable().partitions(); p++) {
raftMgr.stopRaftGroup(partitionRaftGroupName(table.tableId(), p));
}
+
+ table.internalTable().storage().stop();
+ table.internalTable().close();
} catch (Exception e) {
LOG.info("Unable to stop table [name={}]", e, table.name());
}
@@ -570,7 +599,7 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
private CompletableFuture<?> createTableLocally(long causalityToken, String name, UUID tblId, int partitions) {
TableConfiguration tableCfg = tablesCfg.tables().get(name);
- TableStorage tableStorage = dataStorageMgr.engine(tableCfg.dataStorage()).createTable(tableCfg);
+ MvTableStorage tableStorage = dataStorageMgr.engine(tableCfg.dataStorage()).createMvTable(tableCfg);
tableStorage.start();
@@ -1286,19 +1315,6 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
"Only multi version storages are supported. Current storage is a "
+ tbl.internalTable().storage().getClass().getName();
- Supplier<RaftGroupListener> raftGrpLsnrSupplier = () -> new PartitionListener(tblId,
- new VersionedRowStore(
- ((MvTableStorage) tbl.internalTable().storage()).getOrCreateMvPartition(part), txManager));
-
- Supplier<RaftGroupEventsListener> raftGrpEvtsLsnrSupplier = () -> new RebalanceRaftGroupEventsListener(
- metaStorageMgr,
- tblCfg,
- partId,
- part,
- busyLock,
- movePartition(() -> tbl.internalTable().partitionRaftGroupService(part)),
- rebalanceScheduler);
-
// Stable assignments from the meta store, which revision is bounded by the current pending event.
byte[] stableAssignments = metaStorageMgr.get(stablePartAssignmentsKey(partId),
pendingAssignmentsWatchEvent.revision()).join().value();
@@ -1319,14 +1335,34 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
+ " [key={}, partition={}, table={}, localMemberAddress={}]",
pendingAssignmentsWatchEvent.key(), part, tbl.name(), localMember.address());
- raftMgr.startRaftGroupNode(
- partId,
- assignments,
- deltaPeers,
- raftGrpLsnrSupplier,
- raftGrpEvtsLsnrSupplier,
- groupOptionsForInternalTable(tbl.internalTable())
- );
+ if (raftMgr.shouldHaveRaftGroupLocally(deltaPeers)) {
+ MvPartitionStorage partitionStorage = tbl.internalTable().storage().getOrCreateMvPartition(part);
+
+ RaftGroupOptions groupOptions = groupOptionsForPartition(tbl.internalTable(), partitionStorage, assignments);
+
+ RaftGroupListener raftGrpLsnr = new PartitionListener(
+ tblId,
+ new VersionedRowStore(partitionStorage, txManager)
+ );
+
+ RaftGroupEventsListener raftGrpEvtsLsnr = new RebalanceRaftGroupEventsListener(
+ metaStorageMgr,
+ tblCfg,
+ partId,
+ part,
+ busyLock,
+ movePartition(() -> tbl.internalTable().partitionRaftGroupService(part)),
+ rebalanceScheduler
+ );
+
+ raftMgr.startRaftGroupNode(
+ partId,
+ assignments,
+ raftGrpLsnr,
+ raftGrpEvtsLsnr,
+ groupOptions
+ );
+ }
} catch (NodeStoppingException e) {
// no-op
}
@@ -1394,13 +1430,10 @@ public class TableManager extends Producer<TableEvent, TableEventParameters> imp
? Collections.emptyList()
: (List<ClusterNode>) ByteUtils.fromBytes(pendingFromMetastorage);
- List<ClusterNode> appliedPeers = Stream.concat(stableAssignments.stream(), pendingAssignments.stream())
- .collect(Collectors.toList());
-
try {
ClusterNode localMember = raftMgr.server().clusterService().topologyService().localMember();
- if (!appliedPeers.contains(localMember)) {
+ if (!stableAssignments.contains(localMember) && !pendingAssignments.contains(localMember)) {
raftMgr.stopRaftGroup(partId);
}
} catch (NodeStoppingException e) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index 22f9b480b..ee8699231 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -135,6 +135,18 @@ public class PartitionListener implements RaftGroupListener {
return;
}
+ long commandIndex = clo.index();
+
+ long storageAppliedIndex = storage.lastAppliedIndex();
+
+ assert storageAppliedIndex < commandIndex
+ : "Pending write command has a higher index than already processed commands [commandIndex=" + commandIndex
+ + ", storageAppliedIndex=" + storageAppliedIndex + ']';
+
+ // TODO IGNITE-17081 IGNITE-17077
+ // Applied index is set non-atomically. This is a wrong and non-recoverable behavior. Will be fixed later.
+ storage.lastAppliedIndex(commandIndex);
+
if (command instanceof InsertCommand) {
clo.result(handleInsertCommand((InsertCommand) command));
} else if (command instanceof DeleteCommand) {
@@ -404,11 +416,13 @@ public class PartitionListener implements RaftGroupListener {
boolean stateChanged = txManager.changeState(txId, TxState.PENDING, cmd.finish() ? TxState.COMMITED : TxState.ABORTED);
- if (txManager.state(txId) == TxState.COMMITED) {
+ // This code is technically incorrect and assumes that "stateChanged" is always true. This was done because transaction state is not
+ // persisted and thus FinishTxCommand couldn't be completed on recovery after node restart ("changeState" uses "replace").
+ if (/*txManager.state(txId) == TxState.COMMITED*/cmd.finish()) {
cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
storage.commitWrite(ByteBuffer.wrap(key), txId);
});
- } else if (txManager.state(txId) == TxState.ABORTED) {
+ } else /*if (txManager.state(txId) == TxState.ABORTED)*/ {
cmd.lockedKeys().getOrDefault(lockId, new ArrayList<>()).forEach(key -> {
storage.abortWrite(ByteBuffer.wrap(key));
});
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/InitPartitionSnapshotReader.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/InitPartitionSnapshotReader.java
new file mode 100644
index 000000000..d1086986b
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/InitPartitionSnapshotReader.java
@@ -0,0 +1,94 @@
+/*
+ * 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.raft.snapshot;
+
+import java.io.IOException;
+import java.util.Set;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.SnapshotMeta;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Snapshot reader used for raft group bootstrap. Reads initial state of the storage.
+ */
+class InitPartitionSnapshotReader extends SnapshotReader {
+ /** Instance of snapshot storage for shared fields access. */
+ private final PartitionSnapshotStorage snapshotStorage;
+
+ /**
+ * Constructor.
+ *
+ * @param snapshotStorage Snapshot storage.
+ */
+ public InitPartitionSnapshotReader(PartitionSnapshotStorage snapshotStorage) {
+ this.snapshotStorage = snapshotStorage;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotMeta load() {
+ return snapshotStorage.snapshotMeta;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String getPath() {
+ return snapshotStorage.snapshotUri;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Set<String> listFiles() {
+ // No files in the snapshot.
+ return Set.of();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public @Nullable Message getFileMeta(String fileName) {
+ // No files in the snapshot.
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String generateURIForCopy() {
+ //TODO IGNITE-17083
+ throw new UnsupportedOperationException("Not implemented yet: https://issues.apache.org/jira/browse/IGNITE-17083");
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean init(Void opts) {
+ // No-op.
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void shutdown() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void close() throws IOException {
+ // No-op.
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorage.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorage.java
new file mode 100644
index 000000000..61dcfe0cc
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorage.java
@@ -0,0 +1,123 @@
+/*
+ * 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.raft.snapshot;
+
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.SnapshotMeta;
+import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.option.SnapshotCopierOptions;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
+import org.apache.ignite.raft.jraft.storage.SnapshotThrottle;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotCopier;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Snapshot storage for {@link MvPartitionStorage}.
+ *
+ * @see PartitionSnapshotStorageFactory
+ */
+class PartitionSnapshotStorage implements SnapshotStorage {
+ /** Snapshot URI. Points to a snapshot folder. Never created on physical storage. */
+ final String snapshotUri;
+
+ /** Raft options. */
+ final RaftOptions raftOptions;
+
+ /** Instance of partition storage. */
+ final MvPartitionStorage partitionStorage;
+
+ /** Snapshot meta, constructed from the storage data and reaft group configuration. */
+ final SnapshotMeta snapshotMeta;
+
+ /** Snapshot throttle instance. */
+ @Nullable SnapshotThrottle snapshotThrottle;
+
+ /**
+ * Constructor.
+ *
+ * @param snapshotUri Snapshot URI.
+ * @param raftOptions RAFT options.
+ * @param partitionStorage Partition storage.
+ * @param snapshotMeta Snapshot meta.
+ */
+ public PartitionSnapshotStorage(
+ String snapshotUri,
+ RaftOptions raftOptions,
+ MvPartitionStorage partitionStorage,
+ SnapshotMeta snapshotMeta
+ ) {
+ this.snapshotUri = snapshotUri;
+ this.raftOptions = raftOptions;
+ this.partitionStorage = partitionStorage;
+ this.snapshotMeta = snapshotMeta;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean init(Void opts) {
+ // No-op.
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void shutdown() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean setFilterBeforeCopyRemote() {
+ // Option is not supported.
+ return false;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotWriter create() {
+ return new PartitionSnapshotWriter(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotReader open() {
+ return new InitPartitionSnapshotReader(this);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotReader copyFrom(String uri, SnapshotCopierOptions opts) {
+ //TODO IGNITE-17083
+ throw new UnsupportedOperationException("Not implemented yet: https://issues.apache.org/jira/browse/IGNITE-17083");
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotCopier startToCopyFrom(String uri, SnapshotCopierOptions opts) {
+ //TODO IGNITE-17083
+ throw new UnsupportedOperationException("Not implemented yet: https://issues.apache.org/jira/browse/IGNITE-17083");
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void setSnapshotThrottle(SnapshotThrottle snapshotThrottle) {
+ this.snapshotThrottle = snapshotThrottle;
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorageFactory.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorageFactory.java
new file mode 100644
index 000000000..9e747eaae
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotStorageFactory.java
@@ -0,0 +1,85 @@
+/*
+ * 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.raft.snapshot;
+
+import java.util.List;
+import org.apache.ignite.internal.raft.storage.SnapshotStorageFactory;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.raft.jraft.RaftMessagesFactory;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.SnapshotMeta;
+import org.apache.ignite.raft.jraft.option.RaftOptions;
+import org.apache.ignite.raft.jraft.storage.SnapshotStorage;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotReader;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
+
+/**
+ * Snapshot storage factory for {@link MvPartitionStorage}. Utilizes the fact that every partition already stores its latest applied index
+ * and thus can inself be used as its own snapshot.
+ *
+ * <p/>Uses {@link MvPartitionStorage#persistedIndex()} and configuration, passed into constructor, to create a {@link SnapshotMeta} object
+ * in {@link SnapshotReader#load()}.
+ *
+ * <p/>Snapshot writer doesn't allow explicit save of any actual file. {@link SnapshotWriter#saveMeta(SnapshotMeta)} simply returns
+ * {@code true}, and {@link SnapshotWriter#addFile(String)} throws an exception.
+ */
+public class PartitionSnapshotStorageFactory implements SnapshotStorageFactory {
+ /** Partition storage. */
+ private final MvPartitionStorage partitionStorage;
+
+ /** List of peers. */
+ private final List<String> peers;
+
+ /** List of learners. */
+ private final List<String> learners;
+
+ /** RAFT log index read from {@link MvPartitionStorage#persistedIndex()} during factory instantiation. */
+ private final long persistedRaftIndex;
+
+ /**
+ * Constructor.
+ *
+ * @param partitionStorage MV partition storage.
+ * @param peers List of raft group peers to be used in snapshot meta.
+ * @param learners List of raft group learners to be used in snapshot meta.
+ *
+ * @see SnapshotMeta
+ */
+ @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
+ public PartitionSnapshotStorageFactory(MvPartitionStorage partitionStorage, List<String> peers, List<String> learners) {
+ this.partitionStorage = partitionStorage;
+ this.peers = peers;
+ this.learners = learners;
+
+ persistedRaftIndex = partitionStorage.persistedIndex();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public SnapshotStorage createSnapshotStorage(String uri, RaftOptions raftOptions) {
+ SnapshotMeta snapshotMeta = new RaftMessagesFactory().snapshotMeta()
+ .lastIncludedIndex(persistedRaftIndex)
+ // According to the code of org.apache.ignite.raft.jraft.core.NodeImpl.bootstrap, it's "dangerous" to init term with a value
+ // greater than 1. 0 value of persisted index means that the underlying storage is empty.
+ .lastIncludedTerm(persistedRaftIndex > 0 ? 1 : 0)
+ .peersList(peers)
+ .learnersList(learners)
+ .build();
+
+ return new PartitionSnapshotStorage(uri, raftOptions, partitionStorage, snapshotMeta);
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotWriter.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotWriter.java
new file mode 100644
index 000000000..3c0e179c2
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/snapshot/PartitionSnapshotWriter.java
@@ -0,0 +1,105 @@
+/*
+ * 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.raft.snapshot;
+
+import java.util.Set;
+import org.apache.ignite.raft.jraft.entity.RaftOutter.SnapshotMeta;
+import org.apache.ignite.raft.jraft.rpc.Message;
+import org.apache.ignite.raft.jraft.storage.snapshot.SnapshotWriter;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Snapshot writer used for RAFT log truncation.
+ */
+class PartitionSnapshotWriter extends SnapshotWriter {
+ /** Instance of snapshot storage for shared fields access. */
+ private final PartitionSnapshotStorage snapshotStorage;
+
+ /**
+ * Constructor.
+ *
+ * @param snapshotStorage Snapshot storage.
+ */
+ public PartitionSnapshotWriter(PartitionSnapshotStorage snapshotStorage) {
+ this.snapshotStorage = snapshotStorage;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean init(Void opts) {
+ // No-op.
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String getPath() {
+ return snapshotStorage.snapshotUri;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public Set<String> listFiles() {
+ // No files in the snapshot.
+ return Set.of();
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public @Nullable Message getFileMeta(String fileName) {
+ // No files in the snapshot.
+ return null;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean saveMeta(SnapshotMeta meta) {
+ // No-op. Meta is already in the storage.
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean addFile(String fileName, Message fileMeta) {
+ throw new UnsupportedOperationException("Impossible to add file to the snapshot. Operation is not available.");
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean removeFile(String fileName) {
+ throw new UnsupportedOperationException("Impossible to remove file from the snapshot. Operation is not available.");
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void shutdown() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void close() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void close(boolean keepDataOnError) {
+ // No-op.
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 04f3ba4e4..60328079d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -41,7 +41,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.InternalTable;
import org.apache.ignite.internal.table.distributed.command.DeleteAllCommand;
import org.apache.ignite.internal.table.distributed.command.DeleteCommand;
@@ -74,7 +74,6 @@ import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.raft.client.Command;
import org.apache.ignite.raft.client.Peer;
import org.apache.ignite.raft.client.service.RaftGroupService;
-import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -110,7 +109,7 @@ public class InternalTableImpl implements InternalTable {
private final TxManager txManager;
/** Storage for table data. */
- private final TableStorage tableStorage;
+ private final MvTableStorage tableStorage;
/** Mutex for the partition map update. */
public Object updatePartMapMux = new Object();
@@ -133,7 +132,7 @@ public class InternalTableImpl implements InternalTable {
Function<NetworkAddress, String> netAddrResolver,
Function<NetworkAddress, ClusterNode> clusterNodeResolver,
TxManager txManager,
- TableStorage tableStorage
+ MvTableStorage tableStorage
) {
this.tableName = tableName;
this.tableId = tableId;
@@ -147,7 +146,7 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull TableStorage storage() {
+ public MvTableStorage storage() {
return tableStorage;
}
@@ -159,7 +158,7 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull UUID tableId() {
+ public UUID tableId() {
return tableId;
}
@@ -358,7 +357,7 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx) {
+ public Publisher<BinaryRow> scan(int p, @Nullable InternalTransaction tx) {
if (p < 0 || p >= partitions) {
throw new IllegalArgumentException(
IgniteStringFormatter.format(
@@ -391,7 +390,7 @@ public class InternalTableImpl implements InternalTable {
/** {@inheritDoc} */
@Override
- public @NotNull List<String> assignments() {
+ public List<String> assignments() {
awaitLeaderInitialization();
return partitionMap.int2ObjectEntrySet().stream()
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
index bb31f0f61..e25ccdf82 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/VersionedRowStore.java
@@ -21,9 +21,12 @@ import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -69,6 +72,28 @@ public class VersionedRowStore {
public VersionedRowStore(@NotNull MvPartitionStorage storage, @NotNull TxManager txManager) {
this.storage = Objects.requireNonNull(storage);
this.txManager = Objects.requireNonNull(txManager);
+
+ Set<RowId> ids = new HashSet<>();
+
+ storage.forEach((rowId, binaryRow) -> {
+ if (ids.add(rowId)) {
+ primaryIndex.put(binaryRow.keySlice(), rowId);
+ }
+ });
+ }
+
+ /**
+ * Sets the last applied index value.
+ */
+ public void lastAppliedIndex(long appliedIndex) {
+ storage.lastAppliedIndex(appliedIndex);
+ }
+
+ /**
+ * Index of the highest write command applied to the storage. {@code 0} if index is unknown.
+ */
+ public long lastAppliedIndex() {
+ return storage.lastAppliedIndex();
}
/**
@@ -502,7 +527,7 @@ public class VersionedRowStore {
*/
// TODO: IGNITE-16644 Support snapshots.
public CompletionStage<Void> snapshot(Path path) {
- throw new UnsupportedOperationException("Snapshots are not supported yet.");
+ return CompletableFuture.completedFuture(null);
}
/**
@@ -512,7 +537,6 @@ public class VersionedRowStore {
*/
// TODO: IGNITE-16644 Support snapshots.
public void restoreSnapshot(Path path) {
- throw new UnsupportedOperationException("Snapshots are not supported yet.");
}
/**
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java b/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
index 20805f263..3f60a3262 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/Example.java
@@ -21,6 +21,7 @@ import java.math.BigDecimal;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.binary.BinaryObjects;
import org.apache.ignite.internal.schema.Column;
@@ -55,7 +56,7 @@ public class Example {
TxManagerImpl txManager = new TxManagerImpl(null, new HeapLockManager());
return Collections.singletonList(new TableImpl(new DummyInternalTableImpl(new VersionedRowStore(
- new TestMvPartitionStorage(List.of(), 0), txManager), txManager), null));
+ new TestMvPartitionStorage(List.of(), 0), txManager), txManager, new AtomicLong()), null));
}
/**
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/InteropOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/InteropOperationsTest.java
index c4228e785..826f30308 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/InteropOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/InteropOperationsTest.java
@@ -38,6 +38,7 @@ import java.util.BitSet;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeType;
import org.apache.ignite.internal.schema.NativeTypes;
@@ -118,12 +119,19 @@ public class InteropOperationsTest {
TxManager txManager = new TxManagerImpl(clusterService, new HeapLockManager());
txManager.start();
- INT_TABLE = new DummyInternalTableImpl(new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager), txManager);
+ AtomicLong raftIndex = new AtomicLong();
+
+ INT_TABLE = new DummyInternalTableImpl(
+ new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
+ txManager,
+ raftIndex
+ );
+
SchemaRegistry schemaRegistry = new DummySchemaManagerImpl(SCHEMA);
List<PartitionListener> partitionListeners = List.of(INT_TABLE.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
TABLE = new TableImpl(INT_TABLE, schemaRegistry);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueBinaryViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueBinaryViewOperationsTest.java
index 5ad5de932..282c15ebe 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueBinaryViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueBinaryViewOperationsTest.java
@@ -27,6 +27,7 @@ import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypes;
import org.apache.ignite.internal.schema.SchemaDescriptor;
@@ -463,13 +464,17 @@ public class KeyValueBinaryViewOperationsTest {
TxManager txManager = new TxManagerImpl(clusterService, lockManager);
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
- txManager);
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
return new TableImpl(table, new DummySchemaManagerImpl(schema));
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
index 068f8a4ab..d4f61afb8 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsSimpleSchemaTest.java
@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeType;
@@ -718,13 +719,17 @@ public class KeyValueViewOperationsSimpleSchemaTest {
TxManager txManager = new TxManagerImpl(clusterService, lockManager);
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
- txManager);
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
return new TableImpl(table, new DummySchemaManagerImpl(schema));
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
index 3cd72dbbd..32268c1d7 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/KeyValueViewOperationsTest.java
@@ -43,6 +43,7 @@ import java.util.Map;
import java.util.Objects;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypeSpec;
@@ -633,12 +634,17 @@ public class KeyValueViewOperationsTest {
TxManager txManager = new TxManagerImpl(clusterService, new HeapLockManager());
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
- new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager), txManager);
+ new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
Mapper<TestKeyObject> keyMapper = Mapper.of(TestKeyObject.class);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/MessagingServiceTestUtils.java b/modules/table/src/test/java/org/apache/ignite/internal/table/MessagingServiceTestUtils.java
index 0ae9ec73c..a41faf812 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/MessagingServiceTestUtils.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/MessagingServiceTestUtils.java
@@ -28,6 +28,8 @@ import java.util.Iterator;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import org.apache.ignite.internal.tx.TxManager;
@@ -54,7 +56,8 @@ public class MessagingServiceTestUtils {
*/
public static MessagingService mockMessagingService(
TxManager txManager,
- List<PartitionListener> partitionListeners
+ List<PartitionListener> partitionListeners,
+ Function<PartitionListener, AtomicLong> raftIndexFactory
) {
MessagingService messagingService = Mockito.mock(MessagingService.class, RETURNS_DEEP_STUBS);
@@ -72,7 +75,9 @@ public class MessagingServiceTestUtils {
txId, txFinishRequest.commit(), txManager.lockedKeys(txId)
);
- partitionListeners.forEach(partitionListener -> partitionListener.onWrite(iterator(finishTxCommand)));
+ partitionListeners.forEach(partitionListener ->
+ partitionListener.onWrite(iterator(finishTxCommand, raftIndexFactory.apply(partitionListener)))
+ );
if (txFinishRequest.commit()) {
txManager.commitAsync(txId).get();
@@ -88,13 +93,23 @@ public class MessagingServiceTestUtils {
return messagingService;
}
- private static <T extends Command> Iterator<CommandClosure<T>> iterator(T obj) {
+ private static <T extends Command> Iterator<CommandClosure<T>> iterator(T obj, AtomicLong raftIndex) {
+ long commandIndex = raftIndex.incrementAndGet();
+
CommandClosure<T> closure = new CommandClosure<>() {
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
@Override
public T command() {
return obj;
}
+ /** {@inheritDoc} */
@Override
public void result(@Nullable Serializable res) {
// no-op.
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
index 49f2e9231..4f0a401c3 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordBinaryViewOperationsTest.java
@@ -29,6 +29,7 @@ import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import java.util.Collection;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.InvalidTypeException;
import org.apache.ignite.internal.schema.NativeTypes;
@@ -695,13 +696,17 @@ public class RecordBinaryViewOperationsTest {
TxManager txManager = new TxManagerImpl(clusterService, lockManager);
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
- txManager);
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
return new TableImpl(table, new DummySchemaManagerImpl(schema));
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
index 480efc8f0..eec69f714 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/RecordViewOperationsTest.java
@@ -41,6 +41,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.NativeTypeSpec;
@@ -304,12 +305,17 @@ public class RecordViewOperationsTest {
TxManager txManager = new TxManagerImpl(clusterService, new HeapLockManager());
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
- new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager), txManager);
+ new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
Mapper<TestObjectWithAllTypes> recMapper = Mapper.of(TestObjectWithAllTypes.class);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/SchemaValidationTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/SchemaValidationTest.java
index 0f16896d9..dc2dce0ec 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/SchemaValidationTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/SchemaValidationTest.java
@@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.internal.schema.Column;
import org.apache.ignite.internal.schema.InvalidTypeException;
import org.apache.ignite.internal.schema.NativeTypes;
@@ -64,12 +65,17 @@ public class SchemaValidationTest {
TxManagerImpl txManager = new TxManagerImpl(clusterService, new HeapLockManager());
+ AtomicLong raftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
- new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager), txManager);
+ new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
+ txManager,
+ raftIndex
+ );
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, pl -> raftIndex);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
return table;
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/TxLocalTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/TxLocalTest.java
index 099e68454..fafddb0dd 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/TxLocalTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/TxLocalTest.java
@@ -20,6 +20,8 @@ package org.apache.ignite.internal.table;
import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
import org.apache.ignite.internal.storage.basic.TestMvPartitionStorage;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
import org.apache.ignite.internal.table.distributed.storage.VersionedRowStore;
@@ -60,23 +62,31 @@ public class TxLocalTest extends TxAbstractTest {
igniteTransactions = new IgniteTransactionsImpl(txManager);
+ AtomicLong accountsRaftIndex = new AtomicLong();
+
DummyInternalTableImpl table = new DummyInternalTableImpl(
new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
- txManager
+ txManager,
+ accountsRaftIndex
);
accounts = new TableImpl(table, new DummySchemaManagerImpl(ACCOUNTS_SCHEMA));
+ AtomicLong customersRaftIndex = new AtomicLong();
DummyInternalTableImpl table2 = new DummyInternalTableImpl(
new VersionedRowStore(new TestMvPartitionStorage(List.of(), 0), txManager),
- txManager
+ txManager,
+ customersRaftIndex
);
customers = new TableImpl(table2, new DummySchemaManagerImpl(CUSTOMERS_SCHEMA));
List<PartitionListener> partitionListeners = List.of(table.getPartitionListener(), table2.getPartitionListener());
- MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners);
+ Function<PartitionListener, AtomicLong> raftIndexfactory = pl ->
+ pl == table.getPartitionListener() ? accountsRaftIndex : customersRaftIndex;
+
+ MessagingService messagingService = MessagingServiceTestUtils.mockMessagingService(txManager, partitionListeners, raftIndexfactory);
Mockito.when(clusterService.messagingService()).thenReturn(messagingService);
}
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
index 1cde900c7..f7054994f 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/TableManagerTest.java
@@ -280,7 +280,7 @@ public class TableManagerTest extends IgniteAbstractTest {
*/
@Test
public void testPreconfiguredTable() throws Exception {
- when(rm.updateRaftGroup(any(), any(), any(), any(), any(), any())).thenAnswer(mock ->
+ when(rm.startRaftGroupService(any(), any())).thenAnswer(mock ->
CompletableFuture.completedFuture(mock(RaftGroupService.class)));
TableManager tableManager = createTableManager(tblManagerFut, false);
@@ -456,7 +456,7 @@ public class TableManagerTest extends IgniteAbstractTest {
mockManagersAndCreateTable(scmTbl, tblManagerFut);
- verify(rm, times(PARTITIONS)).updateRaftGroup(anyString(), any(), any(), any(), any(), any());
+ verify(rm, times(PARTITIONS)).startRaftGroupService(anyString(), any());
TableManager tableManager = tblManagerFut.join();
@@ -565,7 +565,7 @@ public class TableManagerTest extends IgniteAbstractTest {
CompletableFuture<TableManager> tblManagerFut,
Phaser phaser
) throws Exception {
- when(rm.updateRaftGroup(any(), any(), any(), any(), any(), any())).thenAnswer(mock -> {
+ when(rm.startRaftGroupService(any(), any())).thenAnswer(mock -> {
RaftGroupService raftGrpSrvcMock = mock(RaftGroupService.class);
when(raftGrpSrvcMock.leader()).thenReturn(new Peer(new NetworkAddress("localhost", 47500)));
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 0b908ee77..985147385 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
@@ -33,6 +33,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -94,6 +95,9 @@ public class PartitionCommandListenerTest {
/** Versioned row store. */
private VersionedRowStore versionedRowStore;
+ /** RAFT index. */
+ private AtomicLong raftIndex = new AtomicLong();
+
/**
* Initializes a table listener before tests.
*/
@@ -328,6 +332,8 @@ public class PartitionCommandListenerTest {
List<IgniteBiTuple<Row, UUID>> txs = new ArrayList<>();
commandListener.onWrite(batchIterator(clo -> {
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
doAnswer(invocation -> {
MultiRowsResponse resp = invocation.getArgument(0);
@@ -373,6 +379,8 @@ public class PartitionCommandListenerTest {
List<IgniteBiTuple<Row, UUID>> txs = new ArrayList<>();
commandListener.onWrite(batchIterator(clo -> {
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
doAnswer(invocation -> {
assertNull(invocation.getArgument(0));
@@ -406,6 +414,8 @@ public class PartitionCommandListenerTest {
List<IgniteBiTuple<Row, UUID>> txs = new ArrayList<>();
commandListener.onWrite(batchIterator(clo -> {
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
doAnswer(invocation -> {
MultiRowsResponse resp = invocation.getArgument(0);
@@ -494,6 +504,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new UpsertCommand(row, txId));
doAnswer(invocation -> {
@@ -519,6 +531,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new DeleteCommand(row, txId));
doAnswer(invocation -> {
@@ -584,6 +598,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new InsertCommand(row, txId));
doAnswer(mock -> {
@@ -617,6 +633,8 @@ public class PartitionCommandListenerTest {
txs.add(new IgniteBiTuple<>(row, txId));
}
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new DeleteExactAllCommand(rows, txId));
doAnswer(invocation -> {
@@ -658,6 +676,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i + 1);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new GetAndReplaceCommand(row, txId));
doAnswer(invocation -> {
@@ -692,6 +712,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new GetAndUpsertCommand(row, txId));
doAnswer(invocation -> {
@@ -726,6 +748,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new GetAndDeleteCommand(row, txId));
doAnswer(invocation -> {
@@ -760,6 +784,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i + 1);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new ReplaceIfExistCommand(row, Timestamp.nextVersion().toUuid()));
doAnswer(invocation -> {
@@ -787,6 +813,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i + 1);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new DeleteExactCommand(row, txId));
doAnswer(invocation -> {
@@ -814,6 +842,8 @@ public class PartitionCommandListenerTest {
Row row = getTestRow(i, i);
txs.add(new IgniteBiTuple<>(row, txId));
+ when(clo.index()).thenReturn(raftIndex.incrementAndGet());
+
when(clo.command()).thenReturn(new ReplaceCommand(row, getTestRow(i, i + 1), txId));
doAnswer(invocation -> {
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index b512b831f..4385ef60a 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -27,11 +27,12 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;
import javax.naming.OperationNotSupportedException;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryRowEx;
-import org.apache.ignite.internal.storage.engine.TableStorage;
+import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.table.distributed.command.GetAllCommand;
import org.apache.ignite.internal.table.distributed.command.GetCommand;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
@@ -66,10 +67,10 @@ public class DummyInternalTableImpl extends InternalTableImpl {
* @param store The store.
* @param txManager Transaction manager.
*/
- public DummyInternalTableImpl(VersionedRowStore store, TxManager txManager) {
+ public DummyInternalTableImpl(VersionedRowStore store, TxManager txManager, AtomicLong raftIndex) {
super("test", UUID.randomUUID(),
Int2ObjectMaps.singleton(0, mock(RaftGroupService.class)),
- 1, null, null, txManager, mock(TableStorage.class));
+ 1, null, null, txManager, mock(MvTableStorage.class));
RaftGroupService svc = partitionMap.get(0);
@@ -81,7 +82,9 @@ public class DummyInternalTableImpl extends InternalTableImpl {
invocationClose -> {
Command cmd = invocationClose.getArgument(0);
- CompletableFuture res = new CompletableFuture();
+ long commandIndex = raftIndex.incrementAndGet();
+
+ CompletableFuture<Serializable> res = new CompletableFuture<>();
CompletableFuture<Void> fut = partitionListener.onBeforeApply(cmd);
@@ -109,11 +112,19 @@ public class DummyInternalTableImpl extends InternalTableImpl {
}
} else {
CommandClosure<WriteCommand> clo = new CommandClosure<>() {
+ /** {@inheritDoc} */
+ @Override
+ public long index() {
+ return commandIndex;
+ }
+
+ /** {@inheritDoc} */
@Override
public WriteCommand command() {
return (WriteCommand) cmd;
}
+ /** {@inheritDoc} */
@Override
public void result(@Nullable Serializable r) {
res.complete(r);