You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2024/02/22 15:23:13 UTC
(ignite-3) branch main updated: IGNITE-21560 Add schema sync before apply BuildIndexCommand (#3262)
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 a34d497884 IGNITE-21560 Add schema sync before apply BuildIndexCommand (#3262)
a34d497884 is described below
commit a34d497884256f23e2d54193304f965280861ccb
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Thu Feb 22 18:23:07 2024 +0300
IGNITE-21560 Add schema sync before apply BuildIndexCommand (#3262)
---
.../apache/ignite/internal/table/TableImpl.java | 5 ----
.../distributed/TableIndexStoragesSupplier.java | 12 ++--------
.../distributed/command/BuildIndexCommand.java | 16 ++++++++++++-
.../distributed/index/IndexUpdateHandler.java | 10 --------
.../table/distributed/raft/PartitionListener.java | 27 +++++++--------------
.../replicator/PartitionReplicaListener.java | 10 ++++++--
.../distributed/index/IndexUpdateHandlerTest.java | 8 +------
.../raft/PartitionCommandListenerTest.java | 2 --
.../replication/PartitionReplicaListenerTest.java | 28 ++++++++++++++++------
.../table/impl/DummyInternalTableImpl.java | 11 +--------
10 files changed, 56 insertions(+), 73 deletions(-)
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 6a9c6f631d..37638dcd38 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
@@ -238,11 +238,6 @@ public class TableImpl implements TableViewInternal {
return adapters;
}
-
- @Override
- public void addIndexToWaitIfAbsent(int indexId) {
- addIndexesToWait(indexId);
- }
};
}
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
index 2398aed61b..39c0844846 100644
--- 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
@@ -19,9 +19,8 @@ package org.apache.ignite.internal.table.distributed;
import java.util.Map;
-/**
- * Supplier table index storages.
- */
+/** Supplier table index storages. */
+@FunctionalInterface
public interface TableIndexStoragesSupplier {
/**
* Returns indexes by their ID.
@@ -29,11 +28,4 @@ public interface TableIndexStoragesSupplier {
* <p>Waits for the primary key index and all other registered indexes to be created.
*/
Map<Integer, TableSchemaAwareIndexStorage> get();
-
- /**
- * Adds index creation waits if it hasn't been created yet.
- *
- * @param indexId Index ID.
- */
- void addIndexToWaitIfAbsent(int indexId);
}
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
index 4bf044361f..30e743e1d0 100644
--- 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
@@ -19,13 +19,16 @@ package org.apache.ignite.internal.table.distributed.command;
import java.util.List;
import java.util.UUID;
+import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.network.annotations.Transferable;
+import org.apache.ignite.internal.network.annotations.Transient;
+import org.apache.ignite.internal.network.annotations.WithSetter;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.table.distributed.TableMessageGroup;
/** State machine command to build a table index. */
@Transferable(TableMessageGroup.Commands.BUILD_INDEX)
-public interface BuildIndexCommand extends WriteCommand {
+public interface BuildIndexCommand extends WriteCommand, CatalogVersionAware {
/** Returns index ID. */
int indexId();
@@ -37,4 +40,15 @@ public interface BuildIndexCommand extends WriteCommand {
/** Returns the catalog version in which the index was created. */
int creationCatalogVersion();
+
+ /** Returns the catalog version that is required to build the index, this is the version getting {@link CatalogIndexStatus#BUILDING}. */
+ @Override
+ @Transient
+ @WithSetter
+ int requiredCatalogVersion();
+
+ @Override
+ default void requiredCatalogVersion(int version) {
+ // No-op.
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandler.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandler.java
index 6e8f7905e7..4bec03b744 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandler.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandler.java
@@ -140,9 +140,6 @@ public class IndexUpdateHandler {
* @param nextRowIdToBuild Row ID for which the index needs to be build next, {@code null} means that the index is build.
*/
public void buildIndex(int indexId, Stream<BinaryRowAndRowId> rowStream, @Nullable RowId nextRowIdToBuild) {
- // TODO: IGNITE-19082 Need another way to wait for index creation
- indexes.addIndexToWaitIfAbsent(indexId);
-
TableSchemaAwareIndexStorage index = indexes.get().get(indexId);
rowStream.forEach(binaryRowAndRowId -> {
@@ -160,13 +157,6 @@ public class IndexUpdateHandler {
indexes.get();
}
- /** Waits for the specific index to be created. */
- public void waitForIndex(int indexId) {
- indexes.addIndexToWaitIfAbsent(indexId);
-
- waitIndexes();
- }
-
private Iterable<TableSchemaAwareIndexStorage> indexes(@Nullable List<Integer> indexIds) {
Map<Integer, TableSchemaAwareIndexStorage> indexStorageById = indexes.get();
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 2a3c099af1..d2cb1a5c15 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
@@ -37,11 +37,9 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.function.Consumer;
-import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogService;
-import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.SafeTimeReorderException;
@@ -517,9 +515,6 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
return;
}
- // It's important to not block any thread while being inside of the "runConsistently" section.
- storageUpdateHandler.getIndexUpdateHandler().waitForIndex(cmd.indexId());
-
BuildIndexRowVersionChooser rowVersionChooser = createBuildIndexRowVersionChooser(cmd);
storage.runConsistently(locker -> {
@@ -612,24 +607,18 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
));
}
- // TODO: IGNITE-21560 Add schema sync to BuildIndexCommand#startBuildingCatalogVersion
- // TODO: IGNITE-21560 Skip command if index was removed
private BuildIndexRowVersionChooser createBuildIndexRowVersionChooser(BuildIndexCommand command) {
- int creationCatalogVersion = command.creationCatalogVersion();
- Catalog creationIndexCatalog = catalogService.catalog(creationCatalogVersion);
+ int indexCreationCatalogVersion = command.creationCatalogVersion();
+ Catalog indexCreationCatalog = catalogService.catalog(indexCreationCatalogVersion);
+
+ assert indexCreationCatalog != null : "indexId=" + command.indexId() + ", catalogVersion=" + indexCreationCatalogVersion;
- assert creationIndexCatalog != null : "indexId=" + command.indexId() + ", catalogVersion=" + creationCatalogVersion;
+ int startBuildingIndexCatalogVersion = command.requiredCatalogVersion();
- int latestCatalogVersion = catalogService.latestCatalogVersion();
- Catalog startBuildingIndexCatalog = IntStream.rangeClosed(creationCatalogVersion, latestCatalogVersion)
- .mapToObj(catalogService::catalog)
- .filter(catalog -> catalog.index(command.indexId()).status() == CatalogIndexStatus.BUILDING)
- .findFirst()
- .orElse(null);
+ Catalog startBuildingIndexCatalog = catalogService.catalog(startBuildingIndexCatalogVersion);
- assert startBuildingIndexCatalog != null : format("indexId={}, catalogVersionFrom={}, catalogVersionTo={}",
- command.indexId(), creationCatalogVersion, latestCatalogVersion);
+ assert startBuildingIndexCatalog != null : "indexId=" + command.indexId() + ", catalogVersion=" + startBuildingIndexCatalogVersion;
- return new BuildIndexRowVersionChooser(storage, creationIndexCatalog.time(), startBuildingIndexCatalog.time());
+ return new BuildIndexRowVersionChooser(storage, indexCreationCatalog.time(), startBuildingIndexCatalog.time());
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index ac157fa270..8ee190a25a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -3719,12 +3719,14 @@ public class PartitionReplicaListener implements ReplicaListener {
}
}
- private static BuildIndexCommand toBuildIndexCommand(BuildIndexReplicaRequest request) {
+ private BuildIndexCommand toBuildIndexCommand(BuildIndexReplicaRequest request) {
return MSG_FACTORY.buildIndexCommand()
.indexId(request.indexId())
.rowIds(request.rowIds())
.finish(request.finish())
.creationCatalogVersion(request.creationCatalogVersion())
+ // We are sure that there will be no error here since the primary replica is sent the request to itself.
+ .requiredCatalogVersion(indexStartBuildingCatalogVersion(request))
.build();
}
@@ -3816,8 +3818,12 @@ public class PartitionReplicaListener implements ReplicaListener {
return TableUtils.indexIdsAtRwTxBeginTs(catalogService, txId, tableId());
}
+ private int indexStartBuildingCatalogVersion(BuildIndexReplicaRequest request) {
+ return findStartBuildingIndexCatalogVersion(catalogService, request.indexId(), request.creationCatalogVersion());
+ }
+
private HybridTimestamp indexStartBuildingActivationTs(BuildIndexReplicaRequest request) {
- int catalogVersion = findStartBuildingIndexCatalogVersion(catalogService, request.indexId(), request.creationCatalogVersion());
+ int catalogVersion = indexStartBuildingCatalogVersion(request);
Catalog catalog = catalogService.catalog(catalogVersion);
diff --git a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandlerTest.java b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandlerTest.java
index dfc6abbde1..6cdb12180f 100644
--- a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandlerTest.java
+++ b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/index/IndexUpdateHandlerTest.java
@@ -33,9 +33,7 @@ import org.apache.ignite.internal.table.distributed.TableSchemaAwareIndexStorage
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.junit.jupiter.api.Test;
-/**
- * For {@link IndexUpdateHandler} testing.
- */
+/** For {@link IndexUpdateHandler} testing. */
public class IndexUpdateHandlerTest extends BaseIgniteAbstractTest {
private static final int PARTITION_ID = 0;
@@ -56,8 +54,6 @@ public class IndexUpdateHandlerTest extends BaseIgniteAbstractTest {
indexUpdateHandler.buildIndex(indexId, Stream.of(row0, row1), row1.rowId().increment());
- verify(indexes).addIndexToWaitIfAbsent(indexId);
-
verify(indexStorage).put(row0.binaryRow(), row0.rowId());
verify(indexStorage).put(row1.binaryRow(), row1.rowId());
@@ -70,8 +66,6 @@ public class IndexUpdateHandlerTest extends BaseIgniteAbstractTest {
indexUpdateHandler.buildIndex(indexId, Stream.of(row2), null);
- verify(indexes).addIndexToWaitIfAbsent(indexId);
-
verify(indexStorage).put(row2.binaryRow(), row2.rowId());
verify(indexStorage.storage()).setNextRowIdToBuild(null);
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 5e26174b6c..bd534b2861 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
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.table.distributed.raft;
-import static org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus.BUILDING;
import static org.apache.ignite.internal.util.ArrayUtils.asList;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsInAnyOrder;
@@ -506,7 +505,6 @@ public class PartitionCommandListenerTest extends BaseIgniteAbstractTest {
int indexId = pkStorage.id();
doNothing().when(indexUpdateHandler).buildIndex(eq(indexId), any(Stream.class), any());
- when(indexDescriptor.status()).thenReturn(BUILDING);
RowId row0 = new RowId(PARTITION_ID);
RowId row1 = new RowId(PARTITION_ID);
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 7858d3ccc1..fd362f5700 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
@@ -41,6 +41,7 @@ import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFu
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;
@@ -2805,7 +2806,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
@ParameterizedTest(name = "readOnly = {0}")
@ValueSource(booleans = {true, false})
void testStaleTxOperationAfterIndexStartBuilding(boolean readOnly) {
- fireHashIndexStartBuildingEventForStaleTxOperation(hashIndexStorage.id(), 1);
+ fireHashIndexStartBuildingEventForStaleTxOperation(hashIndexStorage.id(), 1, 2);
UUID txId = newTxId();
long beginTs = beginTimestamp(txId).longValue();
@@ -2833,7 +2834,7 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertFalse(invokeBuildIndexReplicaRequestFuture.isDone());
- fireHashIndexStartBuildingEventForStaleTxOperation(indexId, indexCreationCatalogVersion);
+ fireHashIndexStartBuildingEventForStaleTxOperation(indexId, indexCreationCatalogVersion, indexCreationCatalogVersion + 1);
assertThat(invokeBuildIndexReplicaRequestFuture, willCompleteSuccessfully());
assertThat(invokeBuildIndexReplicaRequestAsync(indexId, indexCreationCatalogVersion), willCompleteSuccessfully());
@@ -2843,11 +2844,14 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
@ValueSource(booleans = {false, true})
void testBuildIndexReplicaRequest(boolean failCmd) {
var continueNotBuildIndexCmdFuture = new CompletableFuture<Void>();
+ var buildIndexCommandFuture = new CompletableFuture<BuildIndexCommand>();
when(mockRaftClient.run(any())).thenAnswer(invocation -> {
Command cmd = invocation.getArgument(0);
if (cmd instanceof BuildIndexCommand) {
+ buildIndexCommandFuture.complete((BuildIndexCommand) cmd);
+
return raftClientFutureClosure.apply(cmd);
}
@@ -2865,13 +2869,14 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
int indexId = hashIndexStorage.id();
int indexCreationCatalogVersion = 1;
+ int indexStartBuildingCatalogVersion = 2;
CompletableFuture<?> invokeBuildIndexReplicaRequestFuture = invokeBuildIndexReplicaRequestAsync(
indexId,
indexCreationCatalogVersion
);
- fireHashIndexStartBuildingEventForStaleTxOperation(indexId, indexCreationCatalogVersion);
+ fireHashIndexStartBuildingEventForStaleTxOperation(indexId, indexCreationCatalogVersion, indexStartBuildingCatalogVersion);
assertFalse(upsertFuture.isDone());
assertFalse(invokeBuildIndexReplicaRequestFuture.isDone());
@@ -2888,12 +2893,23 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
assertThat(invokeBuildIndexReplicaRequestFuture, willCompleteSuccessfully());
- HybridTimestamp startBuildingIndexActivationTs = hybridTimestamp(catalogService.catalog(indexCreationCatalogVersion + 1).time());
+ HybridTimestamp startBuildingIndexActivationTs = hybridTimestamp(catalogService.catalog(indexStartBuildingCatalogVersion).time());
verify(safeTimeClock).waitFor(eq(startBuildingIndexActivationTs));
+
+ assertThat(buildIndexCommandFuture, willCompleteSuccessfully());
+
+ BuildIndexCommand buildIndexCommand = buildIndexCommandFuture.join();
+ assertThat(buildIndexCommand.indexId(), equalTo(indexId));
+ assertThat(buildIndexCommand.creationCatalogVersion(), equalTo(indexCreationCatalogVersion));
+ assertThat(buildIndexCommand.requiredCatalogVersion(), equalTo(indexStartBuildingCatalogVersion));
}
- private void fireHashIndexStartBuildingEventForStaleTxOperation(int indexId, int creationIndexCatalogVersion) {
+ private void fireHashIndexStartBuildingEventForStaleTxOperation(
+ int indexId,
+ int creationIndexCatalogVersion,
+ int startBuildingIndexCatalogVersion
+ ) {
var registeredIndexDescriptor = mock(CatalogHashIndexDescriptor.class);
var buildingIndexDescriptor = mock(CatalogHashIndexDescriptor.class);
@@ -2908,8 +2924,6 @@ public class PartitionReplicaListenerTest extends IgniteAbstractTest {
when(buildingIndexDescriptor.txWaitCatalogVersion()).thenReturn(creationIndexCatalogVersion);
- int startBuildingIndexCatalogVersion = creationIndexCatalogVersion + 1;
-
when(catalogService.index(eq(indexId), eq(creationIndexCatalogVersion))).thenReturn(registeredIndexDescriptor);
when(catalogService.index(eq(indexId), eq(startBuildingIndexCatalogVersion))).thenReturn(buildingIndexDescriptor);
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 10a82973ad..4f9f56c624 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
@@ -511,16 +511,7 @@ public class DummyInternalTableImpl extends InternalTableImpl {
* @param indexes Index storage by ID.
*/
public static TableIndexStoragesSupplier createTableIndexStoragesSupplier(Map<Integer, TableSchemaAwareIndexStorage> indexes) {
- return new TableIndexStoragesSupplier() {
- @Override
- public Map<Integer, TableSchemaAwareIndexStorage> get() {
- return indexes;
- }
-
- @Override
- public void addIndexToWaitIfAbsent(int indexId) {
- }
- };
+ return () -> indexes;
}
/**