You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by rp...@apache.org on 2025/03/19 09:44:06 UTC
(ignite-3) branch main updated: IGNITE-24779 Support index lifecycle for per-zone partitions (#5431)
This is an automated email from the ASF dual-hosted git repository.
rpuch 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 e6655c8ed15 IGNITE-24779 Support index lifecycle for per-zone partitions (#5431)
e6655c8ed15 is described below
commit e6655c8ed15b7133f9588b84da5a6132ef4cf029
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Wed Mar 19 13:44:01 2025 +0400
IGNITE-24779 Support index lifecycle for per-zone partitions (#5431)
* ChangeIndexStatusTaskController now supports primacy tracking for
per-zone partitions
* Do not try to start building an index created with its table
* Improved error handling in IndexBuildController (do not swallow
exceptions, instead pass them to failure handler)
* Tracking local partition set growth in TableManager for per-zone
partitions
* Supported per-zone partitions in the module tests
---
.../ignite/internal/index/ItBuildIndexTest.java | 40 ++++-----
.../index/ChangeIndexStatusTaskController.java | 100 ++++++++++++++++++++-
.../index/ChangeIndexStatusTaskScheduler.java | 2 +-
.../internal/index/IndexBuildController.java | 44 ++++++---
.../internal/index/IndexBuildingManager.java | 8 +-
.../index/ChangeIndexStatusTaskControllerTest.java | 94 +++++++++++++++++--
.../internal/index/IndexBuildControllerTest.java | 4 +-
.../internal/index/TestIndexManagementUtils.java | 4 +-
.../partition/replicator/fixtures/Node.java | 3 +-
.../replicator/ZonePartitionReplicaListener.java | 12 +--
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
.../internal/table/distributed/TableManager.java | 9 +-
.../table/distributed/index/IndexUtils.java | 2 +-
.../replicator/PartitionReplicaListener.java | 3 -
14 files changed, 264 insertions(+), 64 deletions(-)
diff --git a/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java b/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
index 1e135ceecad..37e79ce0eaa 100644
--- a/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
+++ b/modules/index/src/integrationTest/java/org/apache/ignite/internal/index/ItBuildIndexTest.java
@@ -25,6 +25,7 @@ import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapTableViewInternal;
import static org.apache.ignite.internal.catalog.CatalogService.DEFAULT_STORAGE_PROFILE;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
import static org.apache.ignite.internal.sql.engine.util.QueryChecker.containsIndexScan;
import static org.apache.ignite.internal.table.distributed.storage.InternalTableImpl.AWAIT_PRIMARY_REPLICA_TIMEOUT;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
@@ -63,6 +64,7 @@ import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.raft.Command;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
import org.apache.ignite.internal.sql.SqlCommon;
import org.apache.ignite.internal.storage.index.IndexStorage;
@@ -160,7 +162,7 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
createAndPopulateTable(nodes, 1);
- var tableGroupId = new TablePartitionId(tableId(TABLE_NAME), 0);
+ var tableGroupId = replicationGroupId(TABLE_NAME, 0);
IgniteImpl primary = primaryReplica(tableGroupId);
@@ -175,6 +177,20 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
return primary;
}
+ private static ReplicationGroupId replicationGroupId(String tableName, int partitionIndex) {
+ IgniteImpl node = unwrapIgniteImpl(CLUSTER.aliveNode());
+
+ HybridClock clock = node.clock();
+ CatalogManager catalogManager = node.catalogManager();
+
+ CatalogTableDescriptor tableDescriptor = catalogManager.activeCatalog(clock.nowLong()).table(SCHEMA_NAME, tableName);
+
+ assertNotNull(tableDescriptor, String.format("Table %s not found", tableName));
+
+ return enabledColocation() ? new ZonePartitionId(tableDescriptor.zoneId(), partitionIndex)
+ : new TablePartitionId(tableDescriptor.id(), partitionIndex);
+ }
+
private static void changePrimaryReplica(IgniteImpl currentPrimary) throws InterruptedException {
IgniteImpl nextPrimary = CLUSTER.runningNodes()
.map(TestWrappers::unwrapIgniteImpl)
@@ -189,7 +205,7 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
// Let's change the primary replica for partition 0.
NodeUtils.transferPrimary(
CLUSTER.runningNodes().map(TestWrappers::unwrapIgniteImpl).collect(toList()),
- new TablePartitionId(tableId(TABLE_NAME), 0),
+ replicationGroupId(TABLE_NAME, 0),
nextPrimary.name()
);
@@ -307,24 +323,6 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
return indexDescriptor.id();
}
- /**
- * Returns the table ID from the catalog.
- *
- * @param tableName Table name.
- */
- private static int tableId(String tableName) {
- IgniteImpl node = unwrapIgniteImpl(CLUSTER.aliveNode());
-
- HybridClock clock = node.clock();
- CatalogManager catalogManager = node.catalogManager();
-
- CatalogTableDescriptor tableDescriptor = catalogManager.activeCatalog(clock.nowLong()).table(SCHEMA_NAME, tableName);
-
- assertNotNull(tableDescriptor, String.format("Table %s not found", tableName));
-
- return tableDescriptor.id();
- }
-
/**
* Waits for the index to be built on all nodes.
*
@@ -373,7 +371,7 @@ public class ItBuildIndexTest extends BaseSqlIntegrationTest {
HybridTimestamp now = node.clock().now();
for (int partitionId = 0; partitionId < internalTable.partitions(); partitionId++) {
- var tableGroupId = new TablePartitionId(internalTable.tableId(), partitionId);
+ var tableGroupId = replicationGroupId(tableName, partitionId);
CompletableFuture<TokenizedAssignments> assignmentsFuture = placementDriver.getAssignments(tableGroupId, now);
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
index f187b40ff31..95bc6151515 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskController.java
@@ -23,6 +23,7 @@ import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import it.unimi.dsi.fastutil.ints.IntArrayList;
import it.unimi.dsi.fastutil.ints.IntPredicate;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -34,10 +35,15 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexStatus;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.events.CatalogEvent;
import org.apache.ignite.internal.catalog.events.CreateIndexEventParameters;
+import org.apache.ignite.internal.catalog.events.CreateTableEventParameters;
+import org.apache.ignite.internal.catalog.events.DropTableEventParameters;
import org.apache.ignite.internal.catalog.events.RemoveIndexEventParameters;
import org.apache.ignite.internal.catalog.events.StoppingIndexEventParameters;
import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.lowwatermark.LowWatermark;
+import org.apache.ignite.internal.lowwatermark.event.ChangeLowWatermarkEventParameters;
+import org.apache.ignite.internal.lowwatermark.event.LowWatermarkEvent;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.placementdriver.PlacementDriver;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
@@ -46,6 +52,7 @@ import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParam
import org.apache.ignite.internal.replicator.PartitionGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.table.LongPriorityQueue;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
/**
@@ -78,11 +85,19 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
private final ClusterService clusterService;
+ private final LowWatermark lowWatermark;
+
private final ChangeIndexStatusTaskScheduler changeIndexStatusTaskScheduler;
/** Tables IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
private final Set<Integer> localNodeIsPrimaryReplicaForTableIds = ConcurrentHashMap.newKeySet();
+ /** Zone IDs for which the local node is the primary replica for the partition with ID {@code 0}. */
+ private final Set<Integer> localNodeIsPrimaryReplicaForZoneIds = ConcurrentHashMap.newKeySet();
+
+ /** A queue for deferred table destruction events. */
+ private final LongPriorityQueue<DestroyTableEvent> destructionEventsQueue = new LongPriorityQueue<>(DestroyTableEvent::catalogVersion);
+
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
private final AtomicBoolean closeGuard = new AtomicBoolean();
@@ -91,11 +106,13 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
CatalogManager catalogManager,
PlacementDriver placementDriver,
ClusterService clusterService,
+ LowWatermark lowWatermark,
ChangeIndexStatusTaskScheduler changeIndexStatusTaskScheduler
) {
this.catalogService = catalogManager;
this.placementDriver = placementDriver;
this.clusterService = clusterService;
+ this.lowWatermark = lowWatermark;
this.changeIndexStatusTaskScheduler = changeIndexStatusTaskScheduler;
}
@@ -122,6 +139,13 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
catalogService.listen(CatalogEvent.INDEX_REMOVED, EventListener.fromConsumer(this::onIndexRemoved));
+ if (enabledColocation()) {
+ catalogService.listen(CatalogEvent.TABLE_CREATE, EventListener.fromConsumer(this::onTableCreated));
+ catalogService.listen(CatalogEvent.TABLE_DROP, EventListener.fromConsumer(this::onTableDropped));
+
+ lowWatermark.listen(LowWatermarkEvent.LOW_WATERMARK_CHANGED, EventListener.fromConsumer(this::onLwmChanged));
+ }
+
placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, EventListener.fromConsumer(this::onPrimaryReplicaElected));
}
@@ -129,6 +153,12 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
inBusyLock(busyLock, () -> {
CatalogIndexDescriptor indexDescriptor = parameters.indexDescriptor();
+ if (indexDescriptor.isCreatedWithTable()) {
+ // No need to build an index that was created together with its table as its state (empty) already corresponds
+ // to the table's state (which is also empty).
+ return;
+ }
+
if (localNodeIsPrimaryReplicaForTableIds.contains(indexDescriptor.tableId())) {
// Schedule building the index only if the local node is the primary replica for the 0 partition of the table for which the
// index was created.
@@ -157,6 +187,22 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
inBusyLock(busyLock, () -> changeIndexStatusTaskScheduler.stopStartBuildingTask(parameters.indexId()));
}
+ private void onTableCreated(CreateTableEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ CatalogTableDescriptor tableDescriptor = parameters.tableDescriptor();
+
+ if (localNodeIsPrimaryReplicaForZoneIds.contains(tableDescriptor.zoneId())) {
+ localNodeIsPrimaryReplicaForTableIds.add(tableDescriptor.id());
+ }
+ });
+ }
+
+ private void onTableDropped(DropTableEventParameters parameters) {
+ inBusyLock(busyLock, () -> {
+ destructionEventsQueue.enqueue(new DestroyTableEvent(parameters.catalogVersion(), parameters.tableId()));
+ });
+ }
+
private void onPrimaryReplicaElected(PrimaryReplicaEventParameters parameters) {
inBusyLock(busyLock, () -> {
@@ -170,20 +216,33 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
if (isLocalNode(clusterService, parameters.leaseholderId())) {
scheduleTasksOnPrimaryReplicaElectedBusy(primaryReplicaId);
} else {
- scheduleStopTasksOnPrimaryReplicaElected(primaryReplicaId);
+ handlePrimacyLoss(primaryReplicaId);
}
});
}
+ private void onLwmChanged(ChangeLowWatermarkEventParameters parameters) {
+ int earliestVersion = catalogService.activeCatalogVersion(parameters.newLowWatermark().longValue());
+ List<DestroyTableEvent> tablesToDestroy = destructionEventsQueue.drainUpTo(earliestVersion);
+
+ tablesToDestroy.forEach(event -> {
+ localNodeIsPrimaryReplicaForTableIds.remove(event.tableId());
+
+ changeIndexStatusTaskScheduler.stopTasksForTable(event.tableId());
+ });
+ }
+
private void scheduleTasksOnPrimaryReplicaElectedBusy(PartitionGroupId partitionGroupId) {
// It is safe to get the latest version of the catalog because the PRIMARY_REPLICA_ELECTED event is handled on the metastore thread.
Catalog catalog = catalogService.catalog(catalogService.latestCatalogVersion());
IntArrayList tableIds =
getTableIdsForPrimaryReplicaElected(catalog, partitionGroupId, id -> !localNodeIsPrimaryReplicaForTableIds.contains(id));
-
localNodeIsPrimaryReplicaForTableIds.addAll(tableIds);
+ List<Integer> zoneIds = getZoneIdsForPrimaryReplicaElected(partitionGroupId);
+ localNodeIsPrimaryReplicaForZoneIds.addAll(zoneIds);
+
tableIds.forEach(tableId -> {
for (CatalogIndexDescriptor indexDescriptor : catalog.indexes(tableId)) {
switch (indexDescriptor.status()) {
@@ -204,7 +263,7 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
});
}
- private void scheduleStopTasksOnPrimaryReplicaElected(PartitionGroupId partitionGroupId) {
+ private void handlePrimacyLoss(PartitionGroupId partitionGroupId) {
// It is safe to get the latest version of the catalog because the PRIMARY_REPLICA_ELECTED event is handled on the metastore thread.
Catalog catalog = catalogService.catalog(catalogService.latestCatalogVersion());
@@ -212,6 +271,9 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
getTableIdsForPrimaryReplicaElected(catalog, partitionGroupId, localNodeIsPrimaryReplicaForTableIds::contains);
localNodeIsPrimaryReplicaForTableIds.removeAll(tableIds);
+ if (enabledColocation()) {
+ localNodeIsPrimaryReplicaForZoneIds.remove(((ZonePartitionId) partitionGroupId).zoneId());
+ }
tableIds.forEach(changeIndexStatusTaskScheduler::stopTasksForTable);
}
@@ -241,4 +303,36 @@ class ChangeIndexStatusTaskController implements ManuallyCloseable {
return tableIds;
}
+
+ private List<Integer> getZoneIdsForPrimaryReplicaElected(PartitionGroupId partitionGroupId) {
+ if (enabledColocation()) {
+ ZonePartitionId zonePartitionId = (ZonePartitionId) partitionGroupId;
+
+ if (!localNodeIsPrimaryReplicaForZoneIds.contains(zonePartitionId.zoneId())) {
+ return List.of(zonePartitionId.zoneId());
+ } else {
+ return List.of();
+ }
+ } else {
+ return List.of();
+ }
+ }
+
+ private static class DestroyTableEvent {
+ final int catalogVersion;
+ final int tableId;
+
+ private DestroyTableEvent(int catalogVersion, int tableId) {
+ this.catalogVersion = catalogVersion;
+ this.tableId = tableId;
+ }
+
+ int catalogVersion() {
+ return catalogVersion;
+ }
+
+ int tableId() {
+ return tableId;
+ }
+ }
}
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java
index 6f6a939a8ee..14fd666e567 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskScheduler.java
@@ -102,7 +102,7 @@ class ChangeIndexStatusTaskScheduler implements ManuallyCloseable {
* Schedules a task for that will transfer the given index to the {@link CatalogIndexStatus#BUILDING} state.
*/
void scheduleStartBuildingTask(CatalogIndexDescriptor indexDescriptor) {
- assert indexDescriptor.status() == CatalogIndexStatus.REGISTERED;
+ assert indexDescriptor.status() == CatalogIndexStatus.REGISTERED : "Index must be REGISTERED " + indexDescriptor;
LOG.info("Scheduling starting of index building. Index: {}", indexDescriptor);
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
index 125ede70cae..8bb969f4947 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java
@@ -49,6 +49,9 @@ import org.apache.ignite.internal.catalog.events.RemoveIndexEventParameters;
import org.apache.ignite.internal.catalog.events.StartBuildingIndexEventParameters;
import org.apache.ignite.internal.close.ManuallyCloseable;
import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.failure.FailureContext;
+import org.apache.ignite.internal.failure.FailureManager;
+import org.apache.ignite.internal.failure.FailureType;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.network.ClusterService;
@@ -98,6 +101,8 @@ class IndexBuildController implements ManuallyCloseable {
private final ClockService clockService;
+ private final FailureManager failureManager;
+
private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
private final AtomicBoolean closeGuard = new AtomicBoolean();
@@ -113,7 +118,8 @@ class IndexBuildController implements ManuallyCloseable {
CatalogService catalogService,
ClusterService clusterService,
PlacementDriver placementDriver,
- ClockService clockService
+ ClockService clockService,
+ FailureManager failureManager
) {
this.indexBuilder = indexBuilder;
this.indexManager = indexManager;
@@ -121,6 +127,7 @@ class IndexBuildController implements ManuallyCloseable {
this.clusterService = clusterService;
this.placementDriver = placementDriver;
this.clockService = clockService;
+ this.failureManager = failureManager;
}
/** Starts component. */
@@ -147,8 +154,8 @@ class IndexBuildController implements ManuallyCloseable {
placementDriver.listen(PRIMARY_REPLICA_ELECTED, EventListener.fromConsumer(this::onPrimaryReplicaElected));
}
- private CompletableFuture<?> onIndexBuilding(StartBuildingIndexEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onIndexBuilding(StartBuildingIndexEventParameters parameters) {
+ inBusyLockAsync(busyLock, () -> {
Catalog catalog = catalogService.catalog(parameters.catalogVersion());
assert catalog != null : "Failed to find a catalog for the specified version [version=" + parameters.catalogVersion()
@@ -209,26 +216,36 @@ class IndexBuildController implements ManuallyCloseable {
primaryReplicationGroupId,
indexDescriptor,
mvTableStorage,
- replicaMeta)));
+ replicaMeta
+ ))
+ );
startBuildIndexFutures.add(startBuildIndexFuture);
}
}
return CompletableFutures.allOf(startBuildIndexFutures);
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ failureManager.process(new FailureContext(FailureType.CRITICAL_ERROR, ex));
+ }
});
}
- private CompletableFuture<?> onIndexRemoved(RemoveIndexEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onIndexRemoved(RemoveIndexEventParameters parameters) {
+ inBusyLockAsync(busyLock, () -> {
indexBuilder.stopBuildingIndexes(parameters.indexId());
return nullCompletedFuture();
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ failureManager.process(new FailureContext(FailureType.CRITICAL_ERROR, ex));
+ }
});
}
- private CompletableFuture<?> onPrimaryReplicaElected(PrimaryReplicaEventParameters parameters) {
- return inBusyLockAsync(busyLock, () -> {
+ private void onPrimaryReplicaElected(PrimaryReplicaEventParameters parameters) {
+ inBusyLockAsync(busyLock, () -> {
if (isLocalNode(clusterService, parameters.leaseholderId())) {
// TODO https://issues.apache.org/jira/browse/IGNITE-22522
// Need to remove TablePartitionId check here and below.
@@ -265,7 +282,9 @@ class IndexBuildController implements ManuallyCloseable {
tableDescriptor,
primaryReplicaId,
mvTableStorage,
- replicaMeta)));
+ replicaMeta
+ ))
+ );
indexFutures.add(future);
}
@@ -296,6 +315,10 @@ class IndexBuildController implements ManuallyCloseable {
return nullCompletedFuture();
}
+ }).whenComplete((res, ex) -> {
+ if (ex != null) {
+ failureManager.process(new FailureContext(FailureType.CRITICAL_ERROR, ex));
+ }
});
}
@@ -321,7 +344,8 @@ class IndexBuildController implements ManuallyCloseable {
((PartitionGroupId) primaryReplicaId).partitionId(),
indexDescriptor,
mvTableStorage,
- enlistmentConsistencyToken(replicaMeta));
+ enlistmentConsistencyToken(replicaMeta)
+ );
} else if (indexDescriptor.status() == AVAILABLE) {
scheduleBuildIndexAfterDisasterRecovery(
tableDescriptor.zoneId(),
diff --git a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
index 37bc4512718..d3521aad5aa 100644
--- a/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
+++ b/modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildingManager.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.failure.FailureManager;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.lowwatermark.LowWatermark;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
@@ -86,7 +87,8 @@ public class IndexBuildingManager implements IgniteComponent {
ClusterService clusterService,
LogicalTopologyService logicalTopologyService,
ClockService clockService,
- FailureManager failureManager
+ FailureManager failureManager,
+ LowWatermark lowWatermark
) {
this.metaStorageManager = metaStorageManager;
@@ -113,7 +115,8 @@ public class IndexBuildingManager implements IgniteComponent {
catalogManager,
clusterService,
placementDriver,
- clockService
+ clockService,
+ failureManager
);
var indexTaskScheduler = new ChangeIndexStatusTaskScheduler(
@@ -131,6 +134,7 @@ public class IndexBuildingManager implements IgniteComponent {
catalogManager,
placementDriver,
clusterService,
+ lowWatermark,
indexTaskScheduler
);
}
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
index a7f334b7764..4aedeeffd8f 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/ChangeIndexStatusTaskControllerTest.java
@@ -20,13 +20,14 @@ package org.apache.ignite.internal.index;
import static java.util.UUID.randomUUID;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.catalog.CatalogTestUtils.createCatalogManagerWithTestUpdateLog;
+import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestamp;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.COLUMN_NAME;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.INDEX_NAME;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.LOCAL_NODE;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.NODE_NAME;
-import static org.apache.ignite.internal.index.TestIndexManagementUtils.TABLE_NAME;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.createTable;
import static org.apache.ignite.internal.index.TestIndexManagementUtils.newPrimaryReplicaMeta;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.IgniteUtils.closeAllManually;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -39,6 +40,7 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.List;
+import org.apache.ignite.internal.catalog.Catalog;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.commands.MakeIndexAvailableCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
@@ -46,13 +48,20 @@ import org.apache.ignite.internal.catalog.descriptors.CatalogIndexDescriptor;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
+import org.apache.ignite.internal.lowwatermark.TestLowWatermark;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.network.ClusterService;
import org.apache.ignite.internal.network.TopologyService;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.table.TableTestUtils;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.jetbrains.annotations.Nullable;
@@ -66,12 +75,17 @@ import org.mockito.junit.jupiter.MockitoExtension;
/** For {@link ChangeIndexStatusTaskController} testing. */
@ExtendWith(MockitoExtension.class)
public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest {
+ private static final String PRECREATED_TABLE_NAME = "precreated-table";
+ private static final String NOT_PRECREATED_TABLE_NAME = "not-precreated-table";
+
private final HybridClock clock = new HybridClockImpl();
private final TestPlacementDriver placementDriver = new TestPlacementDriver();
private final ClusterService clusterService = createClusterService();
+ private final TestLowWatermark lowWatermark = new TestLowWatermark();
+
private CatalogManager catalogManager;
@Mock
@@ -85,10 +99,14 @@ public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest
assertThat(catalogManager.startAsync(new ComponentContext()), willCompleteSuccessfully());
- createTable(catalogManager, TABLE_NAME, COLUMN_NAME);
+ createTable(catalogManager, PRECREATED_TABLE_NAME, COLUMN_NAME);
taskController = new ChangeIndexStatusTaskController(
- catalogManager, placementDriver, clusterService, changeIndexStatusTaskScheduler
+ catalogManager,
+ placementDriver,
+ clusterService,
+ lowWatermark,
+ changeIndexStatusTaskScheduler
);
taskController.start();
@@ -114,6 +132,15 @@ public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest
verify(changeIndexStatusTaskScheduler).scheduleStartBuildingTask(eq(indexDescriptor()));
}
+ @Test
+ void testDoNotStartTaskOnIndexCreationWithTable() {
+ setPrimaryReplicaLocalNode();
+
+ createTable(catalogManager, NOT_PRECREATED_TABLE_NAME, COLUMN_NAME);
+
+ verify(changeIndexStatusTaskScheduler, never()).scheduleStartBuildingTask(any());
+ }
+
@Test
void testStartTaskOnIndexDropEvent() {
setPrimaryReplicaLocalNode();
@@ -176,14 +203,51 @@ public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest
verify(changeIndexStatusTaskScheduler).stopTasksForTable(eq(tableId()));
}
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void testScheduleStartBuildingTasksOnTableAddedToZoneWhereWeArePrimary() {
+ String tableName = NOT_PRECREATED_TABLE_NAME;
+ String indexName = "index2";
+
+ setPrimaryReplicaLocalNode();
+
+ createTable(catalogManager, tableName, COLUMN_NAME);
+ createIndex(tableName, indexName);
+
+ CatalogIndexDescriptor indexDescriptor = indexDescriptor(indexName);
+
+ verify(changeIndexStatusTaskScheduler).scheduleStartBuildingTask(eq(indexDescriptor));
+ }
+
+ @Test
+ @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+ void testStopAllTableTasksOnTableDropFromZoneWhereWeArePrimary() {
+ int tableId = tableId();
+
+ setPrimaryReplicaLocalNode();
+
+ dropTable();
+ raiseLowWatermarkToCurrentCatalogVersion();
+
+ verify(changeIndexStatusTaskScheduler).stopTasksForTable(eq(tableId));
+ }
+
private void createIndex() {
- TestIndexManagementUtils.createIndex(catalogManager, TABLE_NAME, INDEX_NAME, COLUMN_NAME);
+ createIndex(PRECREATED_TABLE_NAME, INDEX_NAME);
+ }
+
+ private void createIndex(String tableName, String indexName) {
+ TestIndexManagementUtils.createIndex(catalogManager, tableName, indexName, COLUMN_NAME);
}
private void dropIndex() {
TestIndexManagementUtils.dropIndex(catalogManager, INDEX_NAME);
}
+ private void dropTable() {
+ TableTestUtils.dropTable(catalogManager, SqlCommon.DEFAULT_SCHEMA_NAME, PRECREATED_TABLE_NAME);
+ }
+
private void setPrimaryReplicaLocalNode() {
setPrimaryReplica(LOCAL_NODE);
}
@@ -193,19 +257,35 @@ public class ChangeIndexStatusTaskControllerTest extends BaseIgniteAbstractTest
}
private void setPrimaryReplica(ClusterNode clusterNode) {
- TablePartitionId groupId = new TablePartitionId(tableId(), 0);
+ ReplicationGroupId groupId = IgniteSystemProperties.enabledColocation()
+ ? new ZonePartitionId(zoneId(), 0)
+ : new TablePartitionId(tableId(), 0);
ReplicaMeta replicaMeta = newPrimaryReplicaMeta(clusterNode, groupId, HybridTimestamp.MIN_VALUE, HybridTimestamp.MAX_VALUE);
assertThat(placementDriver.setPrimaryReplicaMeta(0, groupId, completedFuture(replicaMeta)), willCompleteSuccessfully());
}
+ private void raiseLowWatermarkToCurrentCatalogVersion() {
+ Catalog currentCatalog = catalogManager.activeCatalog(HybridTimestamp.MAX_VALUE.longValue());
+
+ lowWatermark.updateAndNotify(hybridTimestamp(currentCatalog.time()));
+ }
+
private int tableId() {
- return TestIndexManagementUtils.tableId(catalogManager, TABLE_NAME, clock);
+ return TestIndexManagementUtils.tableId(catalogManager, PRECREATED_TABLE_NAME, clock);
+ }
+
+ private int zoneId() {
+ return TableTestUtils.getZoneIdByTableNameStrict(catalogManager, PRECREATED_TABLE_NAME, clock.nowLong());
}
private CatalogIndexDescriptor indexDescriptor() {
- return TestIndexManagementUtils.indexDescriptor(catalogManager, INDEX_NAME, clock);
+ return indexDescriptor(INDEX_NAME);
+ }
+
+ private CatalogIndexDescriptor indexDescriptor(String indexName) {
+ return TestIndexManagementUtils.indexDescriptor(catalogManager, indexName, clock);
}
private static ClusterService createClusterService() {
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
index 713b66aaea8..081120f655a 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/IndexBuildControllerTest.java
@@ -53,6 +53,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.commands.MakeIndexAvailableCommand;
import org.apache.ignite.internal.catalog.commands.StartBuildingIndexCommand;
+import org.apache.ignite.internal.failure.NoOpFailureManager;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
@@ -120,7 +121,8 @@ public class IndexBuildControllerTest extends BaseIgniteAbstractTest {
catalogManager,
clusterService,
placementDriver,
- clockService
+ clockService,
+ new NoOpFailureManager()
);
indexBuildController.start();
diff --git a/modules/index/src/test/java/org/apache/ignite/internal/index/TestIndexManagementUtils.java b/modules/index/src/test/java/org/apache/ignite/internal/index/TestIndexManagementUtils.java
index dc1db53e930..99b8988dc46 100644
--- a/modules/index/src/test/java/org/apache/ignite/internal/index/TestIndexManagementUtils.java
+++ b/modules/index/src/test/java/org/apache/ignite/internal/index/TestIndexManagementUtils.java
@@ -49,7 +49,7 @@ import org.apache.ignite.internal.metastorage.impl.MetaStorageService;
import org.apache.ignite.internal.network.ClusterNodeImpl;
import org.apache.ignite.internal.placementdriver.ReplicaMeta;
import org.apache.ignite.internal.placementdriver.leases.Lease;
-import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.sql.SqlCommon;
import org.apache.ignite.internal.table.TableTestUtils;
import org.apache.ignite.network.ClusterNode;
@@ -131,7 +131,7 @@ class TestIndexManagementUtils {
static ReplicaMeta newPrimaryReplicaMeta(
ClusterNode clusterNode,
- TablePartitionId replicaGroupId,
+ ReplicationGroupId replicaGroupId,
HybridTimestamp startTime,
HybridTimestamp expirationTime
) {
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
index 56149553fee..b738635e28d 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java
@@ -785,7 +785,8 @@ public class Node {
clusterService,
logicalTopologyService,
clockService,
- failureManager
+ failureManager,
+ lowWatermark
);
systemViewManager = new SystemViewManagerImpl(name, catalogManager);
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
index 4a507440f06..b410b1fa2b4 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionReplicaListener.java
@@ -214,7 +214,7 @@ public class ZonePartitionReplicaListener implements ReplicaListener {
return txCleanupRecoveryRequestHandler.handle((TxCleanupRecoveryRequest) request);
}
- return processZoneReplicaRequest(request, replicaPrimacy, senderId);
+ return processZoneReplicaRequest(request, replicaPrimacy);
}
/**
@@ -241,17 +241,9 @@ public class ZonePartitionReplicaListener implements ReplicaListener {
*
* @param request Request to be processed.
* @param replicaPrimacy Replica primacy information.
- * @param senderId Node sender id.
* @return Future with the result of the processing.
*/
- private CompletableFuture<?> processZoneReplicaRequest(
- ReplicaRequest request,
- ReplicaPrimacy replicaPrimacy,
- UUID senderId
- ) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-24526
- // Need to move the necessary part of PartitionReplicaListener#processRequest request processing here
-
+ private CompletableFuture<?> processZoneReplicaRequest(ReplicaRequest request, ReplicaPrimacy replicaPrimacy) {
if (request instanceof VacuumTxStateReplicaRequest) {
return vacuumTxStateReplicaRequestHandler.handle((VacuumTxStateReplicaRequest) request);
} else if (request instanceof UpdateMinimumActiveTxBeginTimeReplicaRequest) {
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 cc65731d6aa..f0f3d862453 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
@@ -1125,7 +1125,8 @@ public class IgniteImpl implements Ignite {
clusterSvc,
logicalTopologyService,
clockService,
- failureManager
+ failureManager,
+ lowWatermark
);
qryEngine = new SqlQueryProcessor(
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 b13e154412f..0a12ee6cbd6 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
@@ -693,11 +693,18 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
Set<TableImpl> zoneTables = zoneTables(zonePartitionId.zoneId());
- PartitionSet singlePartitionIdSet = PartitionSet.of(zonePartitionId.partitionId());
+ int partitionIndex = zonePartitionId.partitionId();
+ PartitionSet singlePartitionIdSet = PartitionSet.of(partitionIndex);
CompletableFuture<?>[] futures = zoneTables.stream()
.map(tbl -> inBusyLockAsync(busyLock, () -> {
return getOrCreatePartitionStorages(tbl, singlePartitionIdSet)
+ .thenRun(() -> {
+ localPartsByTableId.compute(
+ tbl.tableId(),
+ (tableId, oldPartitionSet) -> extendPartitionSet(oldPartitionSet, partitionIndex)
+ );
+ })
.thenRunAsync(() -> inBusyLock(busyLock, () -> {
lowWatermark.getLowWatermarkSafe(lwm ->
registerIndexesToTable(tbl, catalogService, singlePartitionIdSet, tbl.schemaView(), lwm)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUtils.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUtils.java
index 86d960f1a09..3e8ee7a1eb1 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUtils.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/index/IndexUtils.java
@@ -51,7 +51,7 @@ public class IndexUtils {
PartitionSet partitionSet,
SchemaRegistry schemaRegistry
) {
- var storageIndexDescriptor = StorageIndexDescriptor.create(tableDescriptor, indexDescriptor);
+ StorageIndexDescriptor storageIndexDescriptor = StorageIndexDescriptor.create(tableDescriptor, indexDescriptor);
var tableRowConverter = new TableRowToIndexKeyConverter(
schemaRegistry,
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 848b2e7bc43..677a7145aef 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
@@ -522,9 +522,6 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
}
private CompletableFuture<?> processRequest(ReplicaRequest request, ReplicaPrimacy replicaPrimacy, UUID senderId) {
- // TODO https://issues.apache.org/jira/browse/IGNITE-24526
- // Need to move the necessary part of request processing to ZonePartitionReplicaListener
-
boolean hasSchemaVersion = request instanceof SchemaVersionAwareReplicaRequest;
if (hasSchemaVersion) {