This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 11ee199daa IGNITE-18539 Implement build procedure for new indexes
(#1800)
11ee199daa is described below
commit 11ee199daabd4cdc2c450b4de96e0239190c81dd
Author: Kirill Tkalenko <tkalkir...@yandex.ru>
AuthorDate: Fri Mar 31 14:12:15 2023 +0300
IGNITE-18539 Implement build procedure for new indexes (#1800)
---
modules/index/build.gradle | 3 +
.../apache/ignite/internal/index/IndexBuilder.java | 274 +++++++++++++++++++++
.../apache/ignite/internal/index/IndexManager.java | 19 +-
.../ignite/internal/index/IndexManagerTest.java | 3 +-
.../runner/app/ItIgniteNodeRestartTest.java | 2 +-
.../internal/sql/engine/ItBuildIndexTest.java | 143 +++++++++++
.../org/apache/ignite/internal/app/IgniteImpl.java | 2 +-
.../sql/engine/exec/MockedStructuresTest.java | 2 +-
.../ignite/distributed/ItTablePersistenceTest.java | 9 +-
.../distributed/ItTxDistributedTestSingleNode.java | 11 +-
.../apache/ignite/internal/table/TableImpl.java | 42 +++-
.../table/distributed/StorageUpdateHandler.java | 43 +++-
.../distributed/TableIndexStoragesSupplier.java | 40 +++
.../table/distributed/TableMessageGroup.java | 4 +
.../distributed/command/BuildIndexCommand.java | 50 ++++
.../table/distributed/raft/PartitionListener.java | 32 +++
.../internal/table/distributed/IndexBaseTest.java | 17 +-
.../PartitionGcOnWriteConcurrentTest.java | 9 +-
.../table/distributed/PartitionGcOnWriteTest.java | 10 +-
.../distributed/StorageUpdateHandlerTest.java | 132 ++++++++++
.../raft/PartitionCommandListenerTest.java | 93 +++++--
.../PartitionReplicaListenerIndexLockingTest.java | 3 +-
.../replication/PartitionReplicaListenerTest.java | 3 +-
.../table/impl/DummyInternalTableImpl.java | 24 +-
24 files changed, 909 insertions(+), 61 deletions(-)
diff --git a/modules/index/build.gradle b/modules/index/build.gradle
index c104b76510..33c33510ba 100644
--- a/modules/index/build.gradle
+++ b/modules/index/build.gradle
@@ -26,6 +26,9 @@ dependencies {
implementation project(':ignite-schema')
implementation project(':ignite-table')
implementation project(':ignite-transactions')
+ implementation project(':ignite-storage-api')
+ implementation project(':ignite-network-api')
+ implementation project(':ignite-raft-api')
implementation libs.jetbrains.annotations
testImplementation(testFixtures(project(':ignite-configuration')))
diff --git
a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java
b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java
new file mode 100644
index 0000000000..15364e5298
--- /dev/null
+++
b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuilder.java
@@ -0,0 +1,274 @@
+/*
+ * 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.index;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.stream.Collectors.toList;
+import static
org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.schema.configuration.index.TableIndexView;
+import org.apache.ignite.internal.storage.MvPartitionStorage;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
+import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Class for managing the index building process.
+ */
+class IndexBuilder {
+ private static final IgniteLogger LOG =
Loggers.forClass(IndexBuilder.class);
+
+ /** Batch size of row IDs to build the index. */
+ private static final int BUILD_INDEX_ROW_ID_BATCH_SIZE = 100;
+
+ /** Message factory to create messages - RAFT commands. */
+ private static final TableMessagesFactory TABLE_MESSAGES_FACTORY = new
TableMessagesFactory();
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** Index building executor. */
+ private final ExecutorService buildIndexExecutor;
+
+ IndexBuilder(String nodeName, IgniteSpinBusyLock busyLock, ClusterService
clusterService) {
+ this.busyLock = busyLock;
+ this.clusterService = clusterService;
+
+ int cpus = Runtime.getRuntime().availableProcessors();
+
+ buildIndexExecutor = new ThreadPoolExecutor(
+ cpus,
+ cpus,
+ 30,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<>(),
+ NamedThreadFactory.create(nodeName, "build-index", LOG)
+ );
+ }
+
+ /**
+ * Stops the index builder.
+ */
+ void stop() {
+ shutdownAndAwaitTermination(buildIndexExecutor, 10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Initializes the build of the index.
+ */
+ void startIndexBuild(TableIndexView tableIndexView, TableImpl table) {
+ for (int partitionId = 0; partitionId <
table.internalTable().partitions(); partitionId++) {
+ // TODO: IGNITE-19177 Add assignments check
+ buildIndexExecutor.submit(new BuildIndexTask(table,
tableIndexView, partitionId, null));
+ }
+ }
+
+ /**
+ * Task of building a table index for a partition.
+ *
+ * <p>Only the leader of the raft group will manage the building of the
index. Leader sends batches of row IDs via
+ * {@link BuildIndexCommand}, the next batch will only be send after the
previous batch has been processed.
+ *
+ * <p>Index building itself occurs locally on each node of the raft group
when processing {@link BuildIndexCommand}. This ensures that
+ * the index build process in the raft group is consistent and that the
index build process is restored after restarting the raft group
+ * (not from the beginning).
+ */
+ private class BuildIndexTask implements Runnable {
+ private final TableImpl table;
+
+ private final TableIndexView tableIndexView;
+
+ private final int partitionId;
+
+ /**
+ * ID of the next row to build the index from the previous batch,
{@code null} if it is the first row after the index was created
+ * (both on a live node and after a restore).
+ */
+ private final @Nullable RowId nextRowIdToBuildFromPreviousBatch;
+
+ private BuildIndexTask(
+ TableImpl table,
+ TableIndexView tableIndexView,
+ int partitionId,
+ @Nullable RowId nextRowIdToBuildFromPreviousBatch
+ ) {
+ this.table = table;
+ this.tableIndexView = tableIndexView;
+ this.partitionId = partitionId;
+ this.nextRowIdToBuildFromPreviousBatch =
nextRowIdToBuildFromPreviousBatch;
+ }
+
+ @Override
+ public void run() {
+ if (!busyLock.enterBusy()) {
+ return;
+ }
+
+ try {
+ // At the time of creating the index, we should have already
waited for the table to be created and its raft of clients
+ // (services) to start for all partitions, so there should be
no errors.
+ RaftGroupService raftGroupService =
table.internalTable().partitionRaftGroupService(partitionId);
+
+ raftGroupService
+ // We do not check the presence of nodes in the
topology on purpose, so as not to get into races on
+ // rebalancing, it will be more convenient and
reliable for us to wait for a stable topology with a chosen
+ // leader.
+ .refreshAndGetLeaderWithTerm()
+ .thenComposeAsync(leaderWithTerm -> {
+ if (!busyLock.enterBusy()) {
+ return completedFuture(null);
+ }
+
+ try {
+ // At this point, we have a stable topology,
each node of which has already applied all local updates.
+ if
(!localNodeConsistentId().equals(leaderWithTerm.leader().consistentId())) {
+ // TODO: IGNITE-19053 Must handle the
change of leader
+ // TODO: IGNITE-19053 Add a test to change
the leader even at the start of the task
+ return completedFuture(null);
+ }
+
+ List<RowId> batchRowIds = collectRowIdBatch();
+
+ RowId nextRowId =
getNextRowIdForNextBatch(batchRowIds);
+
+ boolean finish = batchRowIds.size() <
BUILD_INDEX_ROW_ID_BATCH_SIZE || nextRowId == null;
+
+ // TODO: IGNITE-19053 Must handle the change
of leader
+ return
raftGroupService.run(createBuildIndexCommand(batchRowIds, finish))
+ .thenRun(() -> {
+ if (!finish) {
+ assert nextRowId != null :
createCommonTableIndexInfo();
+
+ buildIndexExecutor.submit(
+ new
BuildIndexTask(table, tableIndexView, partitionId, nextRowId)
+ );
+ }
+ });
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }, buildIndexExecutor)
+ .whenComplete((unused, throwable) -> {
+ if (throwable != null) {
+ LOG.error("Index build error: [{}]",
throwable, createCommonTableIndexInfo());
+ }
+ });
+ } catch (Throwable t) {
+ LOG.error("Index build error: [{}]", t,
createCommonTableIndexInfo());
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private boolean isLocalNodeLeader(RaftGroupService raftGroupService) {
+ Peer leader = raftGroupService.leader();
+
+ assert leader != null : "tableId=" + table.tableId() + ",
partitionId=" + partitionId;
+
+ return localNodeConsistentId().equals(leader.consistentId());
+ }
+
+ private List<RowId> createBatchRowIds(RowId lastBuiltRowId, int
batchSize) {
+ MvPartitionStorage mvPartition =
table.internalTable().storage().getMvPartition(partitionId);
+
+ assert mvPartition != null : createCommonTableIndexInfo();
+
+ List<RowId> batch = new ArrayList<>(batchSize);
+
+ for (int i = 0; i < batchSize && lastBuiltRowId != null; i++) {
+ lastBuiltRowId = mvPartition.closestRowId(lastBuiltRowId);
+
+ if (lastBuiltRowId == null) {
+ break;
+ }
+
+ batch.add(lastBuiltRowId);
+
+ lastBuiltRowId = lastBuiltRowId.increment();
+ }
+
+ return batch;
+ }
+
+ private BuildIndexCommand createBuildIndexCommand(List<RowId> rowIds,
boolean finish) {
+ return TABLE_MESSAGES_FACTORY.buildIndexCommand()
+
.tablePartitionId(TABLE_MESSAGES_FACTORY.tablePartitionIdMessage()
+ .tableId(table.tableId())
+ .partitionId(partitionId)
+ .build()
+ )
+ .indexId(tableIndexView.id())
+ .rowIds(rowIds.stream().map(RowId::uuid).collect(toList()))
+ .finish(finish)
+ .build();
+ }
+
+ private String createCommonTableIndexInfo() {
+ return "table=" + table.name() + ", tableId=" + table.tableId()
+ + ", partitionId=" + partitionId
+ + ", index=" + tableIndexView.name() + ", indexId=" +
tableIndexView.id();
+ }
+
+ private String localNodeConsistentId() {
+ return clusterService.topologyService().localMember().name();
+ }
+
+ private @Nullable RowId getNextRowIdForNextBatch(List<RowId> batch) {
+ return batch.isEmpty() ? null : batch.get(batch.size() -
1).increment();
+ }
+
+ private @Nullable List<RowId> collectRowIdBatch() {
+ RowId nextRowIdToBuild;
+
+ if (nextRowIdToBuildFromPreviousBatch == null) {
+ nextRowIdToBuild =
table.internalTable().storage().getOrCreateIndex(partitionId,
tableIndexView.id()).getNextRowIdToBuild();
+
+ if (nextRowIdToBuild == null) {
+ // Index has already been built.
+ return null;
+ }
+ } else {
+ nextRowIdToBuild = nextRowIdToBuildFromPreviousBatch;
+ }
+
+ if (nextRowIdToBuildFromPreviousBatch == null) {
+ LOG.info("Start building the index: [{}]",
createCommonTableIndexInfo());
+ }
+
+ return createBatchRowIds(nextRowIdToBuild,
BUILD_INDEX_ROW_ID_BATCH_SIZE);
+ }
+ }
+}
diff --git
a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
index ed514ba196..4b0caa0ccf 100644
---
a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
+++
b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexManager.java
@@ -67,6 +67,7 @@ import org.apache.ignite.lang.IndexAlreadyExistsException;
import org.apache.ignite.lang.IndexNotFoundException;
import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.lang.TableNotFoundException;
+import org.apache.ignite.network.ClusterService;
import org.jetbrains.annotations.NotNull;
/**
@@ -92,17 +93,29 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
/** Prevents double stopping of the component. */
private final AtomicBoolean stopGuard = new AtomicBoolean();
+ /** Index builder. */
+ private final IndexBuilder indexBuilder;
+
/**
* Constructor.
*
+ * @param nodeName Node name.
* @param tablesCfg Tables and indexes configuration.
* @param schemaManager Schema manager.
* @param tableManager Table manager.
+ * @param clusterService Cluster service.
*/
- public IndexManager(TablesConfiguration tablesCfg, SchemaManager
schemaManager, TableManager tableManager) {
+ public IndexManager(
+ String nodeName,
+ TablesConfiguration tablesCfg,
+ SchemaManager schemaManager,
+ TableManager tableManager,
+ ClusterService clusterService
+ ) {
this.tablesCfg = Objects.requireNonNull(tablesCfg, "tablesCfg");
this.schemaManager = Objects.requireNonNull(schemaManager,
"schemaManager");
this.tableManager = tableManager;
+ this.indexBuilder = new IndexBuilder(nodeName, busyLock,
clusterService);
}
/** {@inheritDoc} */
@@ -159,6 +172,8 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
busyLock.block();
+ indexBuilder.stop();
+
LOG.info("Index manager stopped");
}
@@ -418,6 +433,8 @@ public class IndexManager extends Producer<IndexEvent,
IndexEventParameters> imp
table.pkId(indexId);
}
}
+
+ indexBuilder.startIndexBuild(tableIndexView, table);
});
return allOf(createIndexFuture, fireEventFuture);
diff --git
a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
index 99e8ff769b..0be0aef1ed 100644
---
a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
+++
b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexManagerTest.java
@@ -54,6 +54,7 @@ import
org.apache.ignite.internal.table.distributed.TableManager;
import org.apache.ignite.internal.tx.impl.HeapLockManager;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IndexNotFoundException;
+import org.apache.ignite.network.ClusterService;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -92,7 +93,7 @@ public class IndexManagerTest {
when(schManager.schemaRegistry(anyLong(),
any())).thenReturn(completedFuture(null));
- indexManager = new IndexManager(tablesConfig, schManager,
tableManagerMock);
+ indexManager = new IndexManager("test", tablesConfig, schManager,
tableManagerMock, mock(ClusterService.class));
indexManager.start();
assertThat(
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index b17823bb1c..94ab5b0fd5 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -391,7 +391,7 @@ public class ItIgniteNodeRestartTest extends
IgniteAbstractTest {
topologyAwareRaftGroupServiceFactory
);
- var indexManager = new IndexManager(tablesConfiguration,
schemaManager, tableManager);
+ var indexManager = new IndexManager(name, tablesConfiguration,
schemaManager, tableManager, clusterSvc);
CatalogManager catalogManager = new CatalogServiceImpl(metaStorageMgr);
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java
new file mode 100644
index 0000000000..3ef6863171
--- /dev/null
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItBuildIndexTest.java
@@ -0,0 +1,143 @@
+/*
+ * 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.sql.engine;
+
+import static java.util.stream.Collectors.joining;
+import static
org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan;
+import static
org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.raft.Peer;
+import org.apache.ignite.internal.raft.service.RaftGroupService;
+import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
+import org.apache.ignite.internal.storage.index.IndexStorage;
+import org.apache.ignite.internal.table.InternalTable;
+import org.apache.ignite.internal.table.TableImpl;
+import org.apache.ignite.lang.IgniteStringFormatter;
+import org.apache.ignite.table.Table;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Integration test of index building.
+ */
+public class ItBuildIndexTest extends ClusterPerClassIntegrationTest {
+ private static final String ZONE_NAME = "zone_table";
+
+ private static final String TABLE_NAME = "test_table";
+
+ private static final String INDEX_NAME = "test_index";
+
+ @AfterEach
+ void tearDown() {
+ sql("DROP TABLE IF EXISTS " + TABLE_NAME);
+ }
+
+ @ParameterizedTest
+ @MethodSource("replicas")
+ void testBuildIndexOnStableTopology(int replicas) throws Exception {
+ sql(IgniteStringFormatter.format("CREATE ZONE IF NOT EXISTS {} WITH
REPLICAS={}, PARTITIONS={}",
+ ZONE_NAME, replicas, 2
+ ));
+
+ sql(IgniteStringFormatter.format(
+ "CREATE TABLE {} (i0 INTEGER PRIMARY KEY, i1 INTEGER) WITH
PRIMARY_ZONE='{}'",
+ TABLE_NAME, ZONE_NAME.toUpperCase()
+ ));
+
+ sql(IgniteStringFormatter.format(
+ "INSERT INTO {} VALUES {}",
+ TABLE_NAME, toValuesString(List.of(1, 1), List.of(2, 2),
List.of(3, 3), List.of(4, 4), List.of(5, 5))
+ ));
+
+ sql(IgniteStringFormatter.format("CREATE INDEX {} ON {} (i1)",
INDEX_NAME, TABLE_NAME));
+
+ // TODO: IGNITE-18733
+ waitForIndex(INDEX_NAME);
+
+ waitForIndexBuild(TABLE_NAME, INDEX_NAME);
+
+ assertQuery(IgniteStringFormatter.format("SELECT * FROM {} WHERE i1 >
0", TABLE_NAME))
+ .matches(containsIndexScan("PUBLIC", TABLE_NAME.toUpperCase(),
INDEX_NAME.toUpperCase()))
+ .returns(1, 1)
+ .returns(2, 2)
+ .returns(3, 3)
+ .returns(4, 4)
+ .returns(5, 5)
+ .check();
+ }
+
+ private static int[] replicas() {
+ // TODO: IGNITE-19086 Fix NullPointerException on insertAll
+ // return new int[]{1, 2, 3};
+ return new int[]{1};
+ }
+
+ private static String toValuesString(List<Object>... values) {
+ return Stream.of(values)
+ .peek(Assertions::assertNotNull)
+ .map(objects ->
objects.stream().map(Object::toString).collect(joining(", ", "(", ")")))
+ .collect(joining(", "));
+ }
+
+ private void waitForIndexBuild(String tableName, String indexName) throws
Exception {
+ for (Ignite clusterNode : CLUSTER_NODES) {
+ CompletableFuture<Table> tableFuture =
clusterNode.tables().tableAsync(tableName);
+
+ assertThat(tableFuture, willCompleteSuccessfully());
+
+ TableImpl tableImpl = (TableImpl) tableFuture.join();
+
+ InternalTable internalTable = tableImpl.internalTable();
+
+ UUID indexId = ((IgniteImpl) clusterNode).clusterConfiguration()
+ .getConfiguration(TablesConfiguration.KEY)
+ .indexes()
+ .get(INDEX_NAME.toUpperCase())
+ .id()
+ .value();
+
+ assertNotNull(indexId, "table=" + tableName + ", index=" +
indexName);
+
+ for (int partitionId = 0; partitionId <
internalTable.partitions(); partitionId++) {
+ RaftGroupService raftGroupService =
internalTable.partitionRaftGroupService(partitionId);
+
+ Stream<Peer> allPeers =
Stream.concat(Stream.of(raftGroupService.leader()),
raftGroupService.peers().stream());
+
+ if
(allPeers.map(Peer::consistentId).noneMatch(clusterNode.name()::equals)) {
+ continue;
+ }
+
+ IndexStorage index =
internalTable.storage().getOrCreateIndex(partitionId, indexId);
+
+ assertTrue(waitForCondition(() -> index.getNextRowIdToBuild()
== null, 10, 10_000));
+ }
+ }
+ }
+}
diff --git
a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index dc715c2d12..76087f301e 100644
---
a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++
b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -486,7 +486,7 @@ public class IgniteImpl implements Ignite {
topologyAwareRaftGroupServiceFactory
);
- indexManager = new IndexManager(tablesConfiguration, schemaManager,
distributedTblMgr);
+ indexManager = new IndexManager(name, tablesConfiguration,
schemaManager, distributedTblMgr, clusterSvc);
catalogManager = new CatalogServiceImpl(metaStorageMgr);
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 0ef110270d..06253e8b62 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
@@ -249,7 +249,7 @@ public class MockedStructuresTest extends
IgniteAbstractTest {
tblManager = mockManagers();
- idxManager = new IndexManager(tblsCfg, schemaManager, tblManager);
+ idxManager = new IndexManager(NODE_NAME, tblsCfg, schemaManager,
tblManager, cs);
idxManager.start();
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 7f0977a79a..667016c8a5 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
@@ -82,6 +82,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaL
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.impl.HeapLockManager;
@@ -375,8 +376,12 @@ public class ItTablePersistenceTest extends
ItAbstractListenerSnapshotTest<Parti
PartitionDataStorage partitionDataStorage = new
TestPartitionDataStorage(mvPartitionStorage);
- StorageUpdateHandler storageUpdateHandler =
- new StorageUpdateHandler(0, partitionDataStorage,
Map::of, tableCfg.dataStorage());
+ StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(
+ 0,
+ partitionDataStorage,
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
+ tableCfg.dataStorage()
+ );
PartitionListener listener = new PartitionListener(
partitionDataStorage,
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 93d6e253af..e9e3152a8d 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,6 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
-import java.util.function.Supplier;
import java.util.stream.IntStream;
import org.apache.ignite.internal.affinity.AffinityUtils;
import org.apache.ignite.internal.affinity.Assignment;
@@ -90,6 +89,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaL
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.table.distributed.storage.InternalTableImpl;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.tx.InternalTransaction;
@@ -440,8 +440,13 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
PendingComparableValuesTracker<Long> storageIndexTracker = new
PendingComparableValuesTracker<>(0L);
PartitionDataStorage partitionDataStorage = new
TestPartitionDataStorage(testMpPartStorage);
- Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes = ()
-> Map.of(pkStorage.get().id(), pkStorage.get());
- StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(partId, partitionDataStorage, indexes, dsCfg);
+
+ StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(
+ partId,
+ partitionDataStorage,
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
+ dsCfg
+ );
TopologyAwareRaftGroupServiceFactory
topologyAwareRaftGroupServiceFactory = new TopologyAwareRaftGroupServiceFactory(
clusterServices.get(assignment),
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
index 560c0b0543..1e406f54c6 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/TableImpl.java
@@ -42,6 +42,7 @@ import org.apache.ignite.internal.schema.row.Row;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
import org.apache.ignite.internal.table.distributed.SortedIndexLocker;
+import org.apache.ignite.internal.table.distributed.TableIndexStoragesSupplier;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
import org.apache.ignite.internal.tx.LockManager;
import org.apache.ignite.lang.ErrorGroups;
@@ -222,20 +223,28 @@ public class TableImpl implements Table {
}
/** Returns a supplier of index storage wrapper factories for given
partition. */
- public Supplier<Map<UUID, TableSchemaAwareIndexStorage>>
indexStorageAdapters(int partId) {
- return () -> {
- awaitIndexes();
+ public TableIndexStoragesSupplier indexStorageAdapters(int partId) {
+ return new TableIndexStoragesSupplier() {
+ @Override
+ public Map<UUID, TableSchemaAwareIndexStorage> get() {
+ awaitIndexes();
- List<IndexStorageAdapterFactory> factories = new
ArrayList<>(indexStorageAdapterFactories.values());
+ List<IndexStorageAdapterFactory> factories = new
ArrayList<>(indexStorageAdapterFactories.values());
- Map<UUID, TableSchemaAwareIndexStorage> adapters = new HashMap<>();
+ Map<UUID, TableSchemaAwareIndexStorage> adapters = new
HashMap<>();
- for (IndexStorageAdapterFactory factory : factories) {
- TableSchemaAwareIndexStorage storage = factory.create(partId);
- adapters.put(storage.id(), storage);
+ for (IndexStorageAdapterFactory factory : factories) {
+ TableSchemaAwareIndexStorage storage =
factory.create(partId);
+ adapters.put(storage.id(), storage);
+ }
+
+ return adapters;
}
- return adapters;
+ @Override
+ public void addIndexToWaitIfAbsent(UUID indexId) {
+ addIndexesToWait(List.of(indexId));
+ }
};
}
@@ -385,14 +394,25 @@ public class TableImpl implements Table {
}
/**
- * Adds indexes to wait before inserting data into the table.
+ * Adds indexes to wait, if not already created, before inserting data
into the table.
*
* @param indexIds Indexes Index IDs.
*/
// TODO: IGNITE-19082 Needs to be redone/improved
public void addIndexesToWait(Collection<UUID> indexIds) {
for (UUID indexId : indexIds) {
- indexesToWait.computeIfAbsent(indexId, uuid -> new
CompletableFuture<>());
+ indexesToWait.compute(indexId, (indexId0, awaitIndexFuture) -> {
+ if (awaitIndexFuture != null) {
+ return awaitIndexFuture;
+ }
+
+ if (indexStorageAdapterFactories.containsKey(indexId) &&
indexLockerFactories.containsKey(indexId)) {
+ // Index is already registered and created.
+ return null;
+ }
+
+ return new CompletableFuture<>();
+ });
}
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
index b842038cae..de76d70584 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java
@@ -28,7 +28,6 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
-import java.util.function.Supplier;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.schema.BinaryRow;
import org.apache.ignite.internal.schema.BinaryTuple;
@@ -54,7 +53,7 @@ public class StorageUpdateHandler {
/** Partition storage with access to MV data of a partition. */
private final PartitionDataStorage storage;
- private final Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes;
+ private final TableIndexStoragesSupplier indexes;
/** Last recorded GC low watermark. */
private final AtomicReference<HybridTimestamp> lastRecordedLwm = new
AtomicReference<>();
@@ -73,7 +72,7 @@ public class StorageUpdateHandler {
public StorageUpdateHandler(
int partitionId,
PartitionDataStorage storage,
- Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes,
+ TableIndexStoragesSupplier indexes,
DataStorageConfiguration dsCfg
) {
this.partitionId = partitionId;
@@ -416,4 +415,42 @@ public class StorageUpdateHandler {
public void waitIndexes() {
indexes.get();
}
+
+ /**
+ * Builds an index for all versions of a row.
+ *
+ * <p>Index is expected to exist, skips the tombstones.
+ *
+ * @param indexId Index ID.
+ * @param rowUuids Row uuids.
+ * @param finish Index build completion flag.
+ */
+ public void buildIndex(UUID indexId, List<UUID> rowUuids, boolean finish) {
+ // TODO: IGNITE-19082 Need another way to wait for index creation
+ indexes.addIndexToWaitIfAbsent(indexId);
+
+ TableSchemaAwareIndexStorage index = indexes.get().get(indexId);
+
+ assert index != null : "indexId=" + indexId + ", partitionId=" +
partitionId;
+
+ RowId lastRowId = null;
+
+ for (UUID rowUuid : rowUuids) {
+ lastRowId = new RowId(partitionId, rowUuid);
+
+ try (Cursor<ReadResult> cursor = storage.scanVersions(lastRowId)) {
+ while (cursor.hasNext()) {
+ ReadResult next = cursor.next();
+
+ if (!next.isEmpty()) {
+ index.put(next.binaryRow(), lastRowId);
+ }
+ }
+ }
+ }
+
+ assert lastRowId != null || finish : "indexId=" + indexId + ",
partitionId=" + partitionId;
+
+ index.storage().setNextRowIdToBuild(finish ? null :
lastRowId.increment());
+ }
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableIndexStoragesSupplier.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableIndexStoragesSupplier.java
new file mode 100644
index 0000000000..bd0bb96b49
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableIndexStoragesSupplier.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed;
+
+import java.util.Map;
+import java.util.UUID;
+
+/**
+ * Supplier table index storages.
+ */
+public interface TableIndexStoragesSupplier {
+ /**
+ * Returns indexes by their ID.
+ *
+ * <p>Waits for the primary key index and all other registered indexes to
be created.
+ */
+ Map<UUID, TableSchemaAwareIndexStorage> get();
+
+ /**
+ * Adds index creation waits if it hasn't been created yet.
+ *
+ * @param indexId Index ID.
+ */
+ void addIndexToWaitIfAbsent(UUID indexId);
+}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
index 8a2b1990d1..88373fd5e5 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableMessageGroup.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.table.distributed;
import static
org.apache.ignite.internal.table.distributed.TableMessageGroup.GROUP_TYPE;
+import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
import
org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
import org.apache.ignite.internal.table.distributed.command.TxCleanupCommand;
@@ -152,6 +153,9 @@ public interface TableMessageGroup {
/** Message type for {@link UpdateCommand}. */
short UPDATE = 43;
+ /** Message type for {@link BuildIndexCommand}. */
+ short BUILD_INDEX = 44;
+
/** Message type for {@link TablePartitionIdMessage}. */
short TABLE_PARTITION_ID = 61;
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
new file mode 100644
index 0000000000..dd0527fa4a
--- /dev/null
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/BuildIndexCommand.java
@@ -0,0 +1,50 @@
+/*
+ * 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.command;
+
+import java.util.List;
+import java.util.UUID;
+import org.apache.ignite.internal.raft.WriteCommand;
+import org.apache.ignite.internal.table.distributed.TableMessageGroup;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * State machine command to build a table index.
+ */
+@Transferable(TableMessageGroup.Commands.BUILD_INDEX)
+public interface BuildIndexCommand extends WriteCommand {
+ /**
+ * Returns ID of table partition.
+ */
+ TablePartitionIdMessage tablePartitionId();
+
+ /**
+ * Returns index ID.
+ */
+ UUID indexId();
+
+ /**
+ * Returns row IDs for which to build indexes.
+ */
+ List<UUID> rowIds();
+
+ /**
+ * Returns {@code true} if this batch is the last one.
+ */
+ boolean finish();
+}
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 04c6fb16e8..0954dea8ff 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
@@ -48,6 +48,7 @@ import
org.apache.ignite.internal.storage.PartitionTimestampCursor;
import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
+import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
import
org.apache.ignite.internal.table.distributed.command.TablePartitionIdMessage;
import org.apache.ignite.internal.table.distributed.command.TxCleanupCommand;
@@ -165,6 +166,8 @@ public class PartitionListener implements RaftGroupListener
{
handleTxCleanupCommand((TxCleanupCommand) command,
commandIndex, commandTerm);
} else if (command instanceof SafeTimeSyncCommand) {
handleSafeTimeSyncCommand((SafeTimeSyncCommand) command,
commandIndex, commandTerm);
+ } else if (command instanceof BuildIndexCommand) {
+ handleBuildIndexCommand((BuildIndexCommand) command,
commandIndex, commandTerm);
} else {
assert false : "Command was not found [cmd=" + command +
']';
}
@@ -422,4 +425,33 @@ public class PartitionListener implements
RaftGroupListener {
public MvPartitionStorage getMvStorage() {
return storage.getStorage();
}
+
+ /**
+ * Handler for the {@link BuildIndexCommand}.
+ *
+ * @param cmd Command.
+ * @param commandIndex RAFT index of the command.
+ * @param commandTerm RAFT term of the command.
+ */
+ void handleBuildIndexCommand(BuildIndexCommand cmd, long commandIndex,
long commandTerm) {
+ // Skips the write command because the storage has already executed it.
+ if (commandIndex <= storage.lastAppliedIndex()) {
+ return;
+ }
+
+ storage.runConsistently(() -> {
+ storageUpdateHandler.buildIndex(cmd.indexId(), cmd.rowIds(),
cmd.finish());
+
+ storage.lastApplied(commandIndex, commandTerm);
+
+ return null;
+ });
+
+ if (cmd.finish()) {
+ LOG.info(
+ "Finish building the index: [tableId={}, partitionId={},
indexId={}]",
+ cmd.tablePartitionId().tableId(),
cmd.tablePartitionId().partitionId(), cmd.indexId()
+ );
+ }
+ }
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
index 30ffda06d3..fbb24b636c 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/IndexBaseTest.java
@@ -44,6 +44,7 @@ import
org.apache.ignite.internal.storage.index.SortedIndexDescriptor.SortedInde
import org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.storage.index.impl.TestSortedIndexStorage;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
@@ -118,12 +119,16 @@ public abstract class IndexBaseTest extends
BaseMvStoragesTest {
storage = new TestMvPartitionStorage(PARTITION_ID);
- storageUpdateHandler = new StorageUpdateHandler(PARTITION_ID, new
TestPartitionDataStorage(storage),
- () -> Map.of(
- pkIndexId, pkStorage,
- sortedIndexId, sortedIndexStorage,
- hashIndexId, hashIndexStorage
- ),
+ Map<UUID, TableSchemaAwareIndexStorage> indexes = Map.of(
+ pkIndexId, pkStorage,
+ sortedIndexId, sortedIndexStorage,
+ hashIndexId, hashIndexStorage
+ );
+
+ storageUpdateHandler = new StorageUpdateHandler(
+ PARTITION_ID,
+ new TestPartitionDataStorage(storage),
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(indexes),
dsCfg
);
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
index 2c3997498a..c782bb327e 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteConcurrentTest.java
@@ -28,6 +28,7 @@ import static org.mockito.Mockito.when;
import java.nio.ByteBuffer;
import java.util.Collections;
+import java.util.Map;
import java.util.UUID;
import java.util.function.Consumer;
import org.apache.ignite.distributed.TestPartitionDataStorage;
@@ -41,6 +42,7 @@ import
org.apache.ignite.internal.schema.configuration.storage.DataStorageConfig
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.MvPartitionStorage.WriteClosure;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.testframework.IgniteTestUtils.RunnableX;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -68,7 +70,12 @@ public class PartitionGcOnWriteConcurrentTest {
when(storage.pollForVacuum(any())).thenReturn(null);
- storageUpdateHandler = new StorageUpdateHandler(1, new
TestPartitionDataStorage(storage), Collections::emptyMap, dsCfg);
+ storageUpdateHandler = new StorageUpdateHandler(
+ PARTITION_ID,
+ new TestPartitionDataStorage(storage),
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
+ dsCfg
+ );
}
@ParameterizedTest
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
index 62a3006d77..39721d97b1 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/PartitionGcOnWriteTest.java
@@ -21,8 +21,8 @@ import static java.util.Collections.singletonMap;
import static java.util.stream.Collectors.toList;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import org.apache.ignite.distributed.TestPartitionDataStorage;
import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.storage.ReadResult;
import org.apache.ignite.internal.storage.RowId;
import org.apache.ignite.internal.storage.impl.TestMvPartitionStorage;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.util.Cursor;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
@@ -58,7 +59,12 @@ public class PartitionGcOnWriteTest extends
BaseMvStoragesTest {
void setUp(@InjectConfiguration("mock.gcOnUpdateBatchSize=" +
GC_BATCH_SIZE) DataStorageConfiguration dsCfg) {
storage = new TestMvPartitionStorage(1);
- storageUpdateHandler = new StorageUpdateHandler(1, new
TestPartitionDataStorage(storage), Collections::emptyMap, dsCfg);
+ storageUpdateHandler = new StorageUpdateHandler(
+ 1,
+ new TestPartitionDataStorage(storage),
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of()),
+ dsCfg
+ );
}
@ParameterizedTest
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
new file mode 100644
index 0000000000..69590f7603
--- /dev/null
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandlerTest.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.singletonList;
+import static java.util.stream.Collectors.toList;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.UUID;
+import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
+import org.apache.ignite.internal.schema.BinaryRow;
+import
org.apache.ignite.internal.schema.configuration.storage.DataStorageConfiguration;
+import org.apache.ignite.internal.storage.ReadResult;
+import org.apache.ignite.internal.storage.RowId;
+import org.apache.ignite.internal.storage.index.IndexStorage;
+import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
+import org.apache.ignite.internal.util.Cursor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * For {@link StorageUpdateHandler} testing.
+ */
+@ExtendWith(ConfigurationExtension.class)
+public class StorageUpdateHandlerTest {
+ private static final int PARTITION_ID = 0;
+
+ @InjectConfiguration
+ private DataStorageConfiguration dataStorageConfig;
+
+ private final HybridClock clock = new HybridClockImpl();
+
+ @Test
+ void testBuildIndex() {
+ PartitionDataStorage partitionStorage =
mock(PartitionDataStorage.class);
+
+ TableSchemaAwareIndexStorage indexStorage = createIndexStorage();
+
+ UUID indexId = UUID.randomUUID();
+
+ TableIndexStoragesSupplier indexes =
mock(TableIndexStoragesSupplier.class);
+
+ when(indexes.get()).thenReturn(Map.of(indexId, indexStorage));
+
+ StorageUpdateHandler storageUpdateHandler =
createStorageUpdateHandler(partitionStorage, indexes);
+
+ RowId rowId0 = new RowId(PARTITION_ID);
+ RowId rowId1 = new RowId(PARTITION_ID);
+
+ List<BinaryRow> rowVersions0 = asList(mock(BinaryRow.class), null);
+ List<BinaryRow> rowVersions1 = asList(mock(BinaryRow.class), null);
+
+ setRowVersions(partitionStorage, Map.of(rowId0.uuid(), rowVersions0,
rowId1.uuid(), rowVersions1));
+
+ storageUpdateHandler.buildIndex(indexId, List.of(rowId0.uuid(),
rowId1.uuid()), false);
+
+ verify(indexStorage).put(rowVersions0.get(0), rowId0);
+ verify(indexStorage, never()).put(rowVersions0.get(1), rowId0);
+
+ verify(indexStorage).put(rowVersions1.get(0), rowId1);
+ verify(indexStorage, never()).put(rowVersions1.get(1), rowId1);
+
+ verify(indexStorage.storage()).setNextRowIdToBuild(rowId1.increment());
+ verify(indexes).addIndexToWaitIfAbsent(indexId);
+
+ // Let's check one more batch - it will be the finishing one.
+ RowId rowId2 = new RowId(PARTITION_ID, UUID.randomUUID());
+
+ List<BinaryRow> rowVersions2 = singletonList(mock(BinaryRow.class));
+
+ setRowVersions(partitionStorage, Map.of(rowId2.uuid(), rowVersions2));
+
+ storageUpdateHandler.buildIndex(indexId, List.of(rowId2.uuid()), true);
+
+ verify(indexStorage).put(rowVersions2.get(0), rowId2);
+
+ verify(indexStorage.storage()).setNextRowIdToBuild(null);
+ verify(indexes, times(2)).addIndexToWaitIfAbsent(indexId);
+ }
+
+ private static TableSchemaAwareIndexStorage createIndexStorage() {
+ TableSchemaAwareIndexStorage indexStorage =
mock(TableSchemaAwareIndexStorage.class);
+
+ IndexStorage storage = mock(IndexStorage.class);
+
+ when(indexStorage.storage()).thenReturn(storage);
+
+ return indexStorage;
+ }
+
+ private StorageUpdateHandler
createStorageUpdateHandler(PartitionDataStorage partitionStorage,
TableIndexStoragesSupplier indexes) {
+ return new StorageUpdateHandler(PARTITION_ID, partitionStorage,
indexes, dataStorageConfig);
+ }
+
+ private void setRowVersions(PartitionDataStorage partitionStorage,
Map<UUID, List<BinaryRow>> rowVersions) {
+ for (Entry<UUID, List<BinaryRow>> entry : rowVersions.entrySet()) {
+ RowId rowId = new RowId(PARTITION_ID, entry.getKey());
+
+ List<ReadResult> readResults = entry.getValue().stream()
+ .map(binaryRow -> ReadResult.createFromCommitted(rowId,
binaryRow, clock.now()))
+ .collect(toList());
+
+
when(partitionStorage.scanVersions(rowId)).thenReturn(Cursor.fromIterable(readResults));
+ }
+ }
+}
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 fb847edfbd..216dee1119 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
@@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.mockito.Answers.RETURNS_DEEP_STUBS;
+import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.any;
@@ -50,7 +51,6 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;
-import java.util.function.Supplier;
import org.apache.ignite.distributed.TestPartitionDataStorage;
import org.apache.ignite.internal.TestHybridClock;
import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
@@ -62,7 +62,6 @@ import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.CommittedConfiguration;
-import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.replicator.command.HybridTimestampMessage;
import
org.apache.ignite.internal.replicator.command.SafeTimePropagatingCommand;
import org.apache.ignite.internal.replicator.command.SafeTimeSyncCommand;
@@ -85,10 +84,12 @@ import
org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
import org.apache.ignite.internal.table.distributed.TableMessagesFactory;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
+import org.apache.ignite.internal.table.distributed.command.BuildIndexCommand;
import org.apache.ignite.internal.table.distributed.command.FinishTxCommand;
import org.apache.ignite.internal.table.distributed.command.TxCleanupCommand;
import org.apache.ignite.internal.table.distributed.command.UpdateCommand;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.testframework.WorkDirectory;
import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
import org.apache.ignite.internal.tx.Timestamp;
@@ -128,9 +129,6 @@ public class PartitionCommandListenerTest {
new Column[]{new Column("value", NativeTypes.INT32, false)}
);
- /** Hybrid clock. */
- private static final HybridClock CLOCK = new HybridClockImpl();
-
/** Table command listener. */
private PartitionListener commandListener;
@@ -157,13 +155,13 @@ public class PartitionCommandListenerTest {
private Path workDir;
/** Factory for command messages. */
- private TableMessagesFactory msgFactory = new TableMessagesFactory();
+ private final TableMessagesFactory msgFactory = new TableMessagesFactory();
/** Factory for replica messages. */
- private ReplicaMessagesFactory replicaMessagesFactory = new
ReplicaMessagesFactory();
+ private final ReplicaMessagesFactory replicaMessagesFactory = new
ReplicaMessagesFactory();
/** Hybrid clock. */
- private HybridClock hybridClock;
+ private final HybridClock hybridClock = new HybridClockImpl();
/** Safe time tracker. */
private PendingComparableValuesTracker<HybridTimestamp> safeTimeTracker;
@@ -173,6 +171,8 @@ public class PartitionCommandListenerTest {
private final RaftGroupConfigurationConverter
raftGroupConfigurationConverter = new RaftGroupConfigurationConverter();
+ private StorageUpdateHandler storageUpdateHandler;
+
/**
* Initializes a table listener before tests.
*/
@@ -184,15 +184,14 @@ public class PartitionCommandListenerTest {
when(clusterService.topologyService().localMember().address()).thenReturn(addr);
- ReplicaService replicaService = mock(ReplicaService.class,
RETURNS_DEEP_STUBS);
-
- hybridClock = new HybridClockImpl();
-
safeTimeTracker = new PendingComparableValuesTracker<>(new
HybridTimestamp(1, 0));
- Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes = () ->
Map.of(pkStorage.id(), pkStorage);
-
- StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(0, partitionDataStorage, indexes, dsCfg);
+ storageUpdateHandler = spy(new StorageUpdateHandler(
+ PARTITION_ID,
+ partitionDataStorage,
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.id(),
pkStorage)),
+ dsCfg
+ ));
commandListener = new PartitionListener(
partitionDataStorage,
@@ -281,9 +280,12 @@ public class PartitionCommandListenerTest {
public void
testOnSnapshotSavePropagateLastAppliedIndexAndTerm(@InjectConfiguration
DataStorageConfiguration dsCfg) {
TestPartitionDataStorage partitionDataStorage = new
TestPartitionDataStorage(mvPartitionStorage);
- Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes = () ->
Map.of(pkStorage.id(), pkStorage);
-
- StorageUpdateHandler storageUpdateHandler = new
StorageUpdateHandler(PARTITION_ID, partitionDataStorage, indexes, dsCfg);
+ StorageUpdateHandler storageUpdateHandler = new StorageUpdateHandler(
+ PARTITION_ID,
+ partitionDataStorage,
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.id(),
pkStorage)),
+ dsCfg
+ );
PartitionListener testCommandListener = new PartitionListener(
partitionDataStorage,
@@ -462,6 +464,49 @@ public class PartitionCommandListenerTest {
applySafeTimeCommand(SafeTimeSyncCommand.class, testClock.now());
}
+ @Test
+ void testBuildIndexCommand() {
+ UUID indexId = UUID.randomUUID();
+
+ doNothing().when(storageUpdateHandler).buildIndex(eq(indexId),
any(List.class), anyBoolean());
+
+ List<UUID> rowUuids0 = List.of(UUID.randomUUID());
+ List<UUID> rowUuids1 = List.of(UUID.randomUUID());
+ List<UUID> rowUuids2 = List.of(UUID.randomUUID());
+
+ InOrder inOrder = inOrder(partitionDataStorage, storageUpdateHandler);
+
+
commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId,
rowUuids0, false), 10, 1);
+
+ inOrder.verify(storageUpdateHandler).buildIndex(indexId, rowUuids0,
false);
+ inOrder.verify(partitionDataStorage).lastApplied(10, 1);
+
+
commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId,
rowUuids1, true), 20, 2);
+
+ inOrder.verify(storageUpdateHandler).buildIndex(indexId, rowUuids1,
true);
+ inOrder.verify(partitionDataStorage).lastApplied(20, 2);
+
+ // Let's check that the command with a lower commandIndex than in the
storage will not be executed.
+
commandListener.handleBuildIndexCommand(createBuildIndexCommand(indexId,
rowUuids2, false), 5, 1);
+
+ inOrder.verify(storageUpdateHandler, never()).buildIndex(indexId,
rowUuids2, false);
+ inOrder.verify(partitionDataStorage, never()).lastApplied(5, 1);
+ }
+
+ private BuildIndexCommand createBuildIndexCommand(UUID indexId, List<UUID>
rowUuids, boolean finish) {
+ return msgFactory.buildIndexCommand()
+ .tablePartitionId(
+ msgFactory.tablePartitionIdMessage()
+ .tableId(UUID.randomUUID())
+ .partitionId(PARTITION_ID)
+ .build()
+ )
+ .indexId(indexId)
+ .rowIds(rowUuids)
+ .finish(finish)
+ .build();
+ }
+
private void applySafeTimeCommand(Class<? extends
SafeTimePropagatingCommand> cls, HybridTimestamp timestamp) {
HybridTimestampMessage safeTime = hybridTimestamp(timestamp);
@@ -572,7 +617,7 @@ public class PartitionCommandListenerTest {
rows.put(Timestamp.nextVersion().toUuid(), row.byteBuffer());
}
- HybridTimestamp commitTimestamp = CLOCK.now();
+ HybridTimestamp commitTimestamp = hybridClock.now();
invokeBatchedCommand(msgFactory.updateAllCommand()
.tablePartitionId(
@@ -608,7 +653,7 @@ public class PartitionCommandListenerTest {
rows.put(readRow(row).uuid(), row.byteBuffer());
}
- HybridTimestamp commitTimestamp = CLOCK.now();
+ HybridTimestamp commitTimestamp = hybridClock.now();
invokeBatchedCommand(msgFactory.updateAllCommand()
.tablePartitionId(
@@ -642,7 +687,7 @@ public class PartitionCommandListenerTest {
keyRows.put(readRow(row).uuid(), null);
}
- HybridTimestamp commitTimestamp = CLOCK.now();
+ HybridTimestamp commitTimestamp = hybridClock.now();
invokeBatchedCommand(msgFactory.updateAllCommand()
.tablePartitionId(
@@ -699,7 +744,7 @@ public class PartitionCommandListenerTest {
}).when(clo).result(any());
}));
- HybridTimestamp commitTimestamp = CLOCK.now();
+ HybridTimestamp commitTimestamp = hybridClock.now();
txIds.forEach(txId ->
invokeBatchedCommand(msgFactory.txCleanupCommand()
.txId(txId)
@@ -743,7 +788,7 @@ public class PartitionCommandListenerTest {
}).when(clo).result(any());
}));
- HybridTimestamp commitTimestamp = CLOCK.now();
+ HybridTimestamp commitTimestamp = hybridClock.now();
txIds.forEach(txId ->
invokeBatchedCommand(msgFactory.txCleanupCommand()
.txId(txId)
@@ -818,7 +863,7 @@ public class PartitionCommandListenerTest {
}).when(clo).result(any());
}));
- HybridTimestamp now = CLOCK.now();
+ HybridTimestamp now = hybridClock.now();
txIds.forEach(txId -> invokeBatchedCommand(
msgFactory.txCleanupCommand()
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index 8cde444fe6..99f7a34063 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -69,6 +69,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaL
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.tx.Lock;
@@ -189,7 +190,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
new StorageUpdateHandler(
PART_ID,
new
TestPartitionDataStorage(TEST_MV_PARTITION_STORAGE),
- () -> Map.of(pkStorage.get().id(), pkStorage.get()),
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
dsCfg
),
peer -> true,
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 837307f0b9..ec6b5bc7f2 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -103,6 +103,7 @@ import
org.apache.ignite.internal.table.distributed.replicator.PartitionReplicaL
import org.apache.ignite.internal.table.distributed.replicator.PlacementDriver;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import
org.apache.ignite.internal.table.distributed.replicator.action.RequestType;
+import org.apache.ignite.internal.table.impl.DummyInternalTableImpl;
import org.apache.ignite.internal.table.impl.DummySchemaManagerImpl;
import org.apache.ignite.internal.testframework.IgniteAbstractTest;
import org.apache.ignite.internal.tostring.IgniteToStringInclude;
@@ -336,7 +337,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
new StorageUpdateHandler(
partId,
partitionDataStorage,
- () -> Map.of(pkStorage.get().id(), pkStorage.get()),
+
DummyInternalTableImpl.createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(),
pkStorage.get())),
dsCfg
),
peer -> localNode.name().equals(peer.consistentId()),
diff --git
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index ac641836e7..0d607f4f65 100644
---
a/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++
b/modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.table.impl;
+import static org.junit.jupiter.api.Assertions.fail;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
@@ -29,7 +30,6 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
-import java.util.function.Supplier;
import javax.naming.OperationNotSupportedException;
import org.apache.ignite.configuration.ConfigurationValue;
import org.apache.ignite.distributed.TestPartitionDataStorage;
@@ -62,6 +62,7 @@ import
org.apache.ignite.internal.storage.index.impl.TestHashIndexStorage;
import org.apache.ignite.internal.table.distributed.HashIndexLocker;
import org.apache.ignite.internal.table.distributed.IndexLocker;
import org.apache.ignite.internal.table.distributed.StorageUpdateHandler;
+import org.apache.ignite.internal.table.distributed.TableIndexStoragesSupplier;
import
org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage;
import org.apache.ignite.internal.table.distributed.raft.PartitionDataStorage;
import org.apache.ignite.internal.table.distributed.raft.PartitionListener;
@@ -263,7 +264,7 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
PendingComparableValuesTracker<HybridTimestamp> safeTime = new
PendingComparableValuesTracker<>(new HybridTimestamp(1, 0));
PartitionDataStorage partitionDataStorage = new
TestPartitionDataStorage(mvPartStorage);
- Supplier<Map<UUID, TableSchemaAwareIndexStorage>> indexes = () ->
Map.of(pkStorage.get().id(), pkStorage.get());
+ TableIndexStoragesSupplier indexes =
createTableIndexStoragesSupplier(Map.of(pkStorage.get().id(), pkStorage.get()));
DataStorageConfiguration dsCfg = mock(DataStorageConfiguration.class);
ConfigurationValue<Integer> gcBatchSizeValue =
mock(ConfigurationValue.class);
@@ -398,4 +399,23 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
safeTimeUpdaterThread = null;
}
}
+
+ /**
+ * Returns dummy table index storages supplier.
+ *
+ * @param indexes Index storage by ID.
+ */
+ public static TableIndexStoragesSupplier
createTableIndexStoragesSupplier(Map<UUID, TableSchemaAwareIndexStorage>
indexes) {
+ return new TableIndexStoragesSupplier() {
+ @Override
+ public Map<UUID, TableSchemaAwareIndexStorage> get() {
+ return indexes;
+ }
+
+ @Override
+ public void addIndexToWaitIfAbsent(UUID indexId) {
+ fail("not supported");
+ }
+ };
+ }
}
