You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2025/05/12 14:49:25 UTC
(ignite-3) branch main updated: IGNITE-25105 Support DisasterRecoveryManager functionality in System views for the Colocation track (#5735)
This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 ba41bed4926 IGNITE-25105 Support DisasterRecoveryManager functionality in System views for the Colocation track (#5735)
ba41bed4926 is described below
commit ba41bed4926473a89937a7f77770ac8bbed800cb
Author: Mirza Aliev <al...@gmail.com>
AuthorDate: Mon May 12 18:49:20 2025 +0400
IGNITE-25105 Support DisasterRecoveryManager functionality in System views for the Colocation track (#5735)
---
.../disaster/DisasterRecoveryManager.java | 97 ++++++++++++++++++-
.../disaster/DisasterRecoverySystemViews.java | 91 +++++++++++++++---
.../disaster/ItDisasterRecoveryManagerTest.java | 2 -
.../disaster/ItDisasterRecoverySystemViewTest.java | 43 ++++-----
...ecoveryZonePartitionsStatesSystemViewTest.java} | 105 +++++++++------------
5 files changed, 230 insertions(+), 108 deletions(-)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
index 5c9c5c3f0b2..345a4f9e312 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
@@ -36,8 +36,10 @@ import static org.apache.ignite.internal.partition.replicator.network.disaster.L
import static org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum.HEALTHY;
import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toTablePartitionIdMessage;
import static org.apache.ignite.internal.replicator.message.ReplicaMessageUtils.toZonePartitionIdMessage;
-import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createGlobalPartitionStatesSystemView;
-import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createLocalPartitionStatesSystemView;
+import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createGlobalTablePartitionStatesSystemView;
+import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createGlobalZonePartitionStatesSystemView;
+import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createLocalTablePartitionStatesSystemView;
+import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createLocalZonePartitionStatesSystemView;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.AVAILABLE;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.DEGRADED;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.READ_ONLY;
@@ -272,9 +274,18 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
@Override
public List<SystemView<?>> systemViews() {
+ if (enabledColocation()) {
+ return List.of(
+ createGlobalZonePartitionStatesSystemView(this),
+ createLocalZonePartitionStatesSystemView(this),
+ createGlobalTablePartitionStatesSystemView(this),
+ createLocalTablePartitionStatesSystemView(this)
+ );
+ }
+
return List.of(
- createGlobalPartitionStatesSystemView(this),
- createLocalPartitionStatesSystemView(this)
+ createGlobalTablePartitionStatesSystemView(this),
+ createLocalTablePartitionStatesSystemView(this)
);
}
@@ -710,6 +721,18 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
try {
Catalog catalog = catalogLatestVersion();
+ if (enabledColocation()) {
+ return localPartitionStatesInternal(
+ zoneNames,
+ nodeNames,
+ partitionIds,
+ catalog,
+ zoneState()
+ )
+ .thenApply(res -> zoneStateToTableState(res, catalog))
+ .thenApply(res -> normalizeTableLocal(res, catalog));
+ }
+
return localPartitionStatesInternal(
zoneNames,
nodeNames,
@@ -737,6 +760,19 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
try {
Catalog catalog = catalogLatestVersion();
+ if (enabledColocation()) {
+ return localPartitionStatesInternal(
+ zoneNames,
+ Set.of(),
+ partitionIds,
+ catalog,
+ zoneState()
+ )
+ .thenApply(res -> zoneStateToTableState(res, catalog))
+ .thenApply(res -> normalizeTableLocal(res, catalog))
+ .thenApply(res -> assembleTableGlobal(res, partitionIds, catalog));
+ }
+
return localPartitionStatesInternal(
zoneNames,
Set.of(),
@@ -751,6 +787,59 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
}
}
+ private Map<TablePartitionId, LocalPartitionStateMessageByNode> zoneStateToTableState(
+ Map<ZonePartitionId, LocalPartitionStateMessageByNode> partitionStateMap,
+ Catalog catalog
+ ) {
+ Map<TablePartitionId, LocalPartitionStateMessageByNode> res = new HashMap<>();
+
+ for (Map.Entry<ZonePartitionId, LocalPartitionStateMessageByNode> entry : partitionStateMap.entrySet()) {
+ int zoneId = entry.getKey().zoneId();
+
+ int partitionId = entry.getKey().partitionId();
+
+ LocalPartitionStateMessageByNode zoneLocalPartitionStateMessageByNode = entry.getValue();
+
+ LocalPartitionStateMessageByNode tableLocalPartitionStateMessageByNode = new LocalPartitionStateMessageByNode(new HashMap<>());
+
+ for (CatalogTableDescriptor tableDescriptor : catalog.tables(zoneId)) {
+ TablePartitionId tablePartitionId = new TablePartitionId(tableDescriptor.id(), partitionId);
+
+ for (Map.Entry<String, LocalPartitionStateMessage> nodeEntry : zoneLocalPartitionStateMessageByNode.entrySet()) {
+ String nodeName = nodeEntry.getKey();
+
+ LocalPartitionStateMessage localPartitionStateMessage = nodeEntry.getValue();
+
+ TableViewInternal tableViewInternal = tableManager.cachedTable(tablePartitionId.tableId());
+
+ if (tableViewInternal == null) {
+ continue;
+ }
+
+ MvPartitionStorage partitionStorage = tableViewInternal.internalTable().storage()
+ .getMvPartition(tablePartitionId.partitionId());
+
+ if (partitionStorage == null) {
+ continue;
+ }
+
+ LocalPartitionStateMessage tableLocalPartitionStateMessage =
+ PARTITION_REPLICATION_MESSAGES_FACTORY.localPartitionStateMessage()
+ .partitionId(toTablePartitionIdMessage(REPLICA_MESSAGES_FACTORY, tablePartitionId))
+ .state(localPartitionStateMessage.state())
+ .logIndex(localPartitionStateMessage.logIndex())
+ .estimatedRows(partitionStorage.estimatedSize())
+ .build();
+
+ tableLocalPartitionStateMessageByNode.put(nodeName, tableLocalPartitionStateMessage);
+ }
+ res.put(tablePartitionId, tableLocalPartitionStateMessageByNode);
+ }
+ }
+
+ return res;
+ }
+
static Function<LocalPartitionStateMessage, TablePartitionId> tableState() {
return state -> state.partitionId().asTablePartitionId();
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoverySystemViews.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoverySystemViews.java
index d3148a07106..47c0a53131a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoverySystemViews.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoverySystemViews.java
@@ -35,15 +35,23 @@ import org.apache.ignite.internal.systemview.api.SystemViews;
/** Helper class for disaster recovery system views. */
class DisasterRecoverySystemViews {
- private static final Comparator<GlobalTablePartitionState> GLOBAL_PARTITION_STATE_COMPARATOR =
+ private static final Comparator<GlobalTablePartitionState> GLOBAL_TABLE_PARTITION_STATE_COMPARATOR =
comparing((GlobalTablePartitionState state) -> state.tableName).thenComparingInt(state -> state.partitionId);
- private static final Comparator<SystemViewLocalPartitionState> SYSTEM_VIEW_LOCAL_PARTITION_STATE_COMPARATOR =
- comparing((SystemViewLocalPartitionState state) -> state.state.tableName)
+ private static final Comparator<SystemViewLocalTablePartitionState> SYSTEM_VIEW_LOCAL_PARTITION_STATE_COMPARATOR =
+ comparing((SystemViewLocalTablePartitionState state) -> state.state.tableName)
.thenComparingInt(state -> state.state.partitionId)
.thenComparing(state -> state.nodeName);
- static SystemView<?> createGlobalPartitionStatesSystemView(DisasterRecoveryManager manager) {
+ private static final Comparator<GlobalPartitionState> GLOBAL_ZONE_PARTITION_STATE_COMPARATOR =
+ comparing((GlobalPartitionState state) -> state.zoneName).thenComparingInt(state -> state.partitionId);
+
+ private static final Comparator<SystemViewLocalZonePartitionState> SYSTEM_VIEW_LOCAL_ZONE_PARTITION_STATE_COMPARATOR =
+ comparing((SystemViewLocalZonePartitionState state) -> state.state.zoneName)
+ .thenComparingInt(state -> state.state.partitionId)
+ .thenComparing(state -> state.nodeName);
+
+ static SystemView<?> createGlobalTablePartitionStatesSystemView(DisasterRecoveryManager manager) {
return SystemViews.<GlobalTablePartitionState>clusterViewBuilder()
.name("GLOBAL_PARTITION_STATES")
.addColumn("ZONE_NAME", STRING, state -> state.zoneName)
@@ -58,12 +66,12 @@ class DisasterRecoverySystemViews {
// They are kept for compatibility with 3.0 version, to allow columns being found by their old names.
.addColumn("STATE", STRING, state -> state.state.name())
// End of legacy columns list. New columns must be added below this line.
- .dataProvider(systemViewPublisher(() -> globalPartitionStatesAsync(manager)))
+ .dataProvider(systemViewPublisher(() -> globalTablePartitionStatesAsync(manager)))
.build();
}
- static SystemView<?> createLocalPartitionStatesSystemView(DisasterRecoveryManager manager) {
- return SystemViews.<SystemViewLocalPartitionState>clusterViewBuilder()
+ static SystemView<?> createLocalTablePartitionStatesSystemView(DisasterRecoveryManager manager) {
+ return SystemViews.<SystemViewLocalTablePartitionState>clusterViewBuilder()
.name("LOCAL_PARTITION_STATES")
.addColumn("NODE_NAME", STRING, state -> state.nodeName)
.addColumn("ZONE_NAME", STRING, state -> state.state.zoneName)
@@ -79,7 +87,31 @@ class DisasterRecoverySystemViews {
// They are kept for compatibility with 3.0 version, to allow columns being found by their old names.
.addColumn("STATE", STRING, state -> state.state.state.name())
// End of legacy columns list. New columns must be added below this line.
- .dataProvider(systemViewPublisher(() -> localPartitionStatesAsync(manager)))
+ .dataProvider(systemViewPublisher(() -> localTablePartitionStatesAsync(manager)))
+ .build();
+ }
+
+ static SystemView<?> createGlobalZonePartitionStatesSystemView(DisasterRecoveryManager manager) {
+ return SystemViews.<GlobalPartitionState>clusterViewBuilder()
+ .name("GLOBAL_ZONE_PARTITION_STATES")
+ .addColumn("ZONE_NAME", STRING, state -> state.zoneName)
+ .addColumn("PARTITION_ID", INT32, state -> state.partitionId)
+ .addColumn("PARTITION_STATE", STRING, state -> state.state.name())
+ .addColumn("ZONE_ID", INT32, state -> state.zoneId)
+ .dataProvider(systemViewPublisher(() -> globalZonePartitionStatesAsync(manager)))
+ .build();
+ }
+
+ static SystemView<?> createLocalZonePartitionStatesSystemView(DisasterRecoveryManager manager) {
+ return SystemViews.<SystemViewLocalZonePartitionState>clusterViewBuilder()
+ .name("LOCAL_ZONE_PARTITION_STATES")
+ .addColumn("NODE_NAME", STRING, state -> state.nodeName)
+ .addColumn("ZONE_NAME", STRING, state -> state.state.zoneName)
+ .addColumn("PARTITION_ID", INT32, state -> state.state.partitionId)
+ .addColumn("PARTITION_STATE", STRING, state -> state.state.state.name())
+ .addColumn("ESTIMATED_ROWS", INT64, state -> state.state.estimatedRows)
+ .addColumn("ZONE_ID", INT32, state -> state.state.zoneId)
+ .dataProvider(systemViewPublisher(() -> localZonePartitionStatesAsync(manager)))
.build();
}
@@ -91,28 +123,59 @@ class DisasterRecoverySystemViews {
};
}
- private static CompletableFuture<Iterator<GlobalTablePartitionState>> globalPartitionStatesAsync(DisasterRecoveryManager manager) {
+ private static CompletableFuture<Iterator<GlobalTablePartitionState>> globalTablePartitionStatesAsync(DisasterRecoveryManager manager) {
return manager.globalTablePartitionStates(Set.of(), Set.of()).thenApply(states -> states.values().stream()
- .sorted(GLOBAL_PARTITION_STATE_COMPARATOR)
+ .sorted(GLOBAL_TABLE_PARTITION_STATE_COMPARATOR)
.iterator()
);
}
- private static CompletableFuture<Iterator<SystemViewLocalPartitionState>> localPartitionStatesAsync(DisasterRecoveryManager manager) {
+ private static CompletableFuture<Iterator<SystemViewLocalTablePartitionState>> localTablePartitionStatesAsync(
+ DisasterRecoveryManager manager
+ ) {
return manager.localTablePartitionStates(Set.of(), Set.of(), Set.of()).thenApply(states -> states.values().stream()
.flatMap(statesByNodeName -> statesByNodeName.entrySet().stream())
- .map(nodeStates -> new SystemViewLocalPartitionState(nodeStates.getKey(), nodeStates.getValue()))
+ .map(nodeStates -> new SystemViewLocalTablePartitionState(nodeStates.getKey(), nodeStates.getValue()))
.sorted(SYSTEM_VIEW_LOCAL_PARTITION_STATE_COMPARATOR)
.iterator()
);
}
- private static class SystemViewLocalPartitionState {
+ private static CompletableFuture<Iterator<GlobalPartitionState>> globalZonePartitionStatesAsync(DisasterRecoveryManager manager) {
+ return manager.globalPartitionStates(Set.of(), Set.of()).thenApply(states -> states.values().stream()
+ .sorted(GLOBAL_ZONE_PARTITION_STATE_COMPARATOR)
+ .iterator()
+ );
+ }
+
+ private static CompletableFuture<Iterator<SystemViewLocalZonePartitionState>> localZonePartitionStatesAsync(
+ DisasterRecoveryManager manager
+ ) {
+ return manager.localPartitionStates(Set.of(), Set.of(), Set.of()).thenApply(states -> states.values().stream()
+ .flatMap(statesByNodeName -> statesByNodeName.entrySet().stream())
+ .map(nodeStates -> new SystemViewLocalZonePartitionState(nodeStates.getKey(), nodeStates.getValue()))
+ .sorted(SYSTEM_VIEW_LOCAL_ZONE_PARTITION_STATE_COMPARATOR)
+ .iterator()
+ );
+ }
+
+ private static class SystemViewLocalTablePartitionState {
private final String nodeName;
private final LocalTablePartitionState state;
- private SystemViewLocalPartitionState(String nodeName, LocalTablePartitionState state) {
+ private SystemViewLocalTablePartitionState(String nodeName, LocalTablePartitionState state) {
+ this.nodeName = nodeName;
+ this.state = state;
+ }
+ }
+
+ private static class SystemViewLocalZonePartitionState {
+ private final String nodeName;
+
+ private final LocalPartitionState state;
+
+ private SystemViewLocalZonePartitionState(String nodeName, LocalPartitionState state) {
this.nodeName = nodeName;
this.state = state;
}
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
index db6c33d1488..109151400c8 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryManagerTest.java
@@ -157,7 +157,6 @@ public class ItDisasterRecoveryManagerTest extends ClusterPerTestIntegrationTest
}
@Test
- @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "false")
@ZoneParams(nodes = 2, replicas = 2, partitions = 2)
void testLocalPartitionStateTable() throws Exception {
IgniteImpl node = unwrapIgniteImpl(cluster.aliveNode());
@@ -236,7 +235,6 @@ public class ItDisasterRecoveryManagerTest extends ClusterPerTestIntegrationTest
}
@Test
- @WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "false")
@ZoneParams(nodes = 2, replicas = 2, partitions = 2)
void testGlobalPartitionStateTable() throws Exception {
IgniteImpl node = unwrapIgniteImpl(cluster.aliveNode());
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java
index 2266276707c..8e3cec3ac82 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java
@@ -19,8 +19,8 @@ package org.apache.ignite.internal.disaster;
import static java.util.stream.Collectors.toList;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
-import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
-import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
+import static org.apache.ignite.internal.disaster.ItDisasterRecoveryZonePartitionsStatesSystemViewTest.estimatedSize;
+import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
import static org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum.HEALTHY;
import static org.apache.ignite.internal.sql.SqlCommon.DEFAULT_SCHEMA_NAME;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.AVAILABLE;
@@ -31,7 +31,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.List;
-import java.util.Objects;
import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.app.IgniteImpl;
@@ -40,17 +39,13 @@ import org.apache.ignite.internal.raft.service.RaftGroupService;
import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
-import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.table.TableImpl;
import org.apache.ignite.internal.table.distributed.PublicApiThreadingTable;
-import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.sql.ColumnType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
/** For integration testing of disaster recovery system views. */
-// TODO https://issues.apache.org/jira/browse/IGNITE-25105
-@WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "false")
public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
/** Table name. */
public static final String TABLE_NAME = "TEST_TABLE";
@@ -118,7 +113,11 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ if (enabledColocation()) {
+ ItDisasterRecoveryZonePartitionsStatesSystemViewTest.waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
+ } else {
+ waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ }
int tableId = getTableId(DEFAULT_SCHEMA_NAME, TABLE_NAME);
@@ -136,7 +135,11 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ if (enabledColocation()) {
+ ItDisasterRecoveryZonePartitionsStatesSystemViewTest.waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
+ } else {
+ waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ }
List<String> nodeNames = CLUSTER.runningNodes().map(Ignite::name).sorted().collect(toList());
@@ -161,7 +164,11 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ if (enabledColocation()) {
+ ItDisasterRecoveryZonePartitionsStatesSystemViewTest.waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
+ } else {
+ waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ }
insertPeople(
TABLE_NAME,
@@ -176,7 +183,7 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
// Small wait is specially added so that the follower can execute the replicated "insert" command and the counter is honestly
// increased.
assertTrue(waitForCondition(
- () -> nodeNames.stream().allMatch(nodeName -> estimatedSize(nodeName, TABLE_NAME, 0) >= 2L),
+ () -> nodeNames.stream().allMatch(nodeName -> estimatedSize(nodeName, TABLE_NAME, 0, CLUSTER) >= 2L),
10,
1_000
));
@@ -226,18 +233,4 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
return catalogManager.catalog(catalogManager.latestCatalogVersion()).table(schemaName, tableName).id();
}
-
- private static long estimatedSize(String nodeName, String tableName, int partitionId) {
- return CLUSTER.runningNodes()
- .filter(ignite -> nodeName.equals(ignite.name()))
- .map(ignite -> {
- TableImpl table = unwrapTableImpl(ignite.tables().table(tableName));
-
- return table.internalTable().storage().getMvPartition(partitionId);
- })
- .filter(Objects::nonNull)
- .map(MvPartitionStorage::estimatedSize)
- .findAny()
- .orElse(-1L);
- }
}
diff --git a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryZonePartitionsStatesSystemViewTest.java
similarity index 61%
copy from modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java
copy to modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryZonePartitionsStatesSystemViewTest.java
index 2266276707c..83415a0f985 100644
--- a/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoverySystemViewTest.java
+++ b/modules/transactions/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryZonePartitionsStatesSystemViewTest.java
@@ -22,7 +22,6 @@ import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
import static org.apache.ignite.internal.TestWrappers.unwrapTableImpl;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
import static org.apache.ignite.internal.partition.replicator.network.disaster.LocalPartitionStateEnum.HEALTHY;
-import static org.apache.ignite.internal.sql.SqlCommon.DEFAULT_SCHEMA_NAME;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.AVAILABLE;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
@@ -34,24 +33,23 @@ import java.util.List;
import java.util.Objects;
import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
+import org.apache.ignite.internal.Cluster;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.raft.service.RaftGroupService;
-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.engine.util.MetadataMatcher;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.table.TableImpl;
-import org.apache.ignite.internal.table.distributed.PublicApiThreadingTable;
import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.sql.ColumnType;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
/** For integration testing of disaster recovery system views. */
-// TODO https://issues.apache.org/jira/browse/IGNITE-25105
-@WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "false")
-public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
+@WithSystemProperty(key = COLOCATION_FEATURE_FLAG, value = "true")
+public class ItDisasterRecoveryZonePartitionsStatesSystemViewTest extends BaseSqlIntegrationTest {
/** Table name. */
public static final String TABLE_NAME = "TEST_TABLE";
@@ -70,61 +68,43 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
@Test
public void globalPatitionStatesMetadata() {
- assertQuery("SELECT * FROM SYSTEM.GLOBAL_PARTITION_STATES")
+ assertQuery("SELECT * FROM SYSTEM.GLOBAL_ZONE_PARTITION_STATES")
.columnMetadata(
new MetadataMatcher().name("ZONE_NAME").type(ColumnType.STRING).nullable(true),
- new MetadataMatcher().name("TABLE_ID").type(ColumnType.INT32).nullable(true),
- new MetadataMatcher().name("SCHEMA_NAME").type(ColumnType.STRING).nullable(true),
- new MetadataMatcher().name("TABLE_NAME").type(ColumnType.STRING).nullable(true),
new MetadataMatcher().name("PARTITION_ID").type(ColumnType.INT32).nullable(true),
new MetadataMatcher().name("PARTITION_STATE").type(ColumnType.STRING).nullable(true),
- new MetadataMatcher().name("ZONE_ID").type(ColumnType.INT32).nullable(true),
- new MetadataMatcher().name("SCHEMA_ID").type(ColumnType.INT32).nullable(true),
- // Legacy columns.
- new MetadataMatcher().name("STATE").type(ColumnType.STRING).nullable(true)
+ new MetadataMatcher().name("ZONE_ID").type(ColumnType.INT32).nullable(true)
)
.check();
}
@Test
public void localPatitionStatesMetadata() {
- assertQuery("SELECT * FROM SYSTEM.LOCAL_PARTITION_STATES")
+ assertQuery("SELECT * FROM SYSTEM.LOCAL_ZONE_PARTITION_STATES")
.columnMetadata(
new MetadataMatcher().name("NODE_NAME").type(ColumnType.STRING).nullable(true),
new MetadataMatcher().name("ZONE_NAME").type(ColumnType.STRING).nullable(true),
- new MetadataMatcher().name("TABLE_ID").type(ColumnType.INT32).nullable(true),
- new MetadataMatcher().name("SCHEMA_NAME").type(ColumnType.STRING).nullable(true),
- new MetadataMatcher().name("TABLE_NAME").type(ColumnType.STRING).nullable(true),
new MetadataMatcher().name("PARTITION_ID").type(ColumnType.INT32).nullable(true),
new MetadataMatcher().name("PARTITION_STATE").type(ColumnType.STRING).nullable(true),
new MetadataMatcher().name("ESTIMATED_ROWS").type(ColumnType.INT64).nullable(true),
- new MetadataMatcher().name("ZONE_ID").type(ColumnType.INT32).nullable(true),
- new MetadataMatcher().name("SCHEMA_ID").type(ColumnType.INT32).nullable(true),
- // Legacy columns.
- new MetadataMatcher().name("STATE").type(ColumnType.STRING).nullable(true)
+ new MetadataMatcher().name("ZONE_ID").type(ColumnType.INT32).nullable(true)
)
.check();
}
- @Test
- void testNoZonesAndTables() {
- assertQuery(globalPartitionStatesSystemViewSql()).returnNothing().check();
- assertQuery(localPartitionStatesSystemViewSql()).returnNothing().check();
- }
-
@Test
void testGlobalPartitionStatesSystemView() {
int partitionsCount = 2;
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
- int tableId = getTableId(DEFAULT_SCHEMA_NAME, TABLE_NAME);
+ int zoneId = getZoneId(ZONE_NAME);
- assertQuery(globalPartitionStatesSystemViewSql())
- .returns(ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 0, AVAILABLE.name())
- .returns(ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 1, AVAILABLE.name())
+ assertQuery(globalZonePartitionStatesSystemViewSql())
+ .returns(ZONE_NAME, 0, AVAILABLE.name(), zoneId)
+ .returns(ZONE_NAME, 1, AVAILABLE.name(), zoneId)
.check();
}
@@ -136,20 +116,20 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
List<String> nodeNames = CLUSTER.runningNodes().map(Ignite::name).sorted().collect(toList());
String nodeName0 = nodeNames.get(0);
String nodeName1 = nodeNames.get(1);
- int tableId = getTableId(DEFAULT_SCHEMA_NAME, TABLE_NAME);
+ int zoneId = getZoneId(ZONE_NAME);
- assertQuery(localPartitionStatesSystemViewSql())
- .returns(nodeName0, ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 0, HEALTHY.name(), 0L)
- .returns(nodeName0, ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 1, HEALTHY.name(), 0L)
- .returns(nodeName1, ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 0, HEALTHY.name(), 0L)
- .returns(nodeName1, ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 1, HEALTHY.name(), 0L)
+ assertQuery(localZonePartitionStatesSystemViewSql())
+ .returns(nodeName0, ZONE_NAME, 0, HEALTHY.name(), 0L, zoneId)
+ .returns(nodeName0, ZONE_NAME, 1, HEALTHY.name(), 0L, zoneId)
+ .returns(nodeName1, ZONE_NAME, 0, HEALTHY.name(), 0L, zoneId)
+ .returns(nodeName1, ZONE_NAME, 1, HEALTHY.name(), 0L, zoneId)
.check();
}
@@ -161,7 +141,7 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), partitionsCount);
- waitLeaderOnAllPartitions(TABLE_NAME, partitionsCount);
+ waitLeaderOnAllPartitions(ZONE_NAME, partitionsCount);
insertPeople(
TABLE_NAME,
@@ -171,64 +151,63 @@ public class ItDisasterRecoverySystemViewTest extends BaseSqlIntegrationTest {
List<String> nodeNames = CLUSTER.runningNodes().map(Ignite::name).sorted().collect(toList());
- int tableId = getTableId(DEFAULT_SCHEMA_NAME, TABLE_NAME);
-
// Small wait is specially added so that the follower can execute the replicated "insert" command and the counter is honestly
// increased.
assertTrue(waitForCondition(
- () -> nodeNames.stream().allMatch(nodeName -> estimatedSize(nodeName, TABLE_NAME, 0) >= 2L),
+ () -> nodeNames.stream().allMatch(nodeName -> estimatedSize(nodeName, TABLE_NAME, 0, CLUSTER) >= 2L),
10,
1_000
));
- assertQuery(localPartitionStatesSystemViewSql())
- .returns(nodeNames.get(0), ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 0, HEALTHY.name(), 2L)
- .returns(nodeNames.get(1), ZONE_NAME, tableId, DEFAULT_SCHEMA_NAME, TABLE_NAME, 0, HEALTHY.name(), 2L)
+ int zoneId = getZoneId(ZONE_NAME);
+
+ assertQuery(localZonePartitionStatesSystemViewSql())
+ .returns(nodeNames.get(0), ZONE_NAME, 0, HEALTHY.name(), 2L, zoneId)
+ .returns(nodeNames.get(1), ZONE_NAME, 0, HEALTHY.name(), 2L, zoneId)
.check();
}
/**
- * waiting a leader for all partitions because later we expect that partitions will be in AVAILABLE state. Without it there won't be
+ * Waiting a leader for all partitions because later we expect that partitions will be in AVAILABLE state. Without it there won't be
* log updating (see {@code LocalPartitionStateEnumWithLogIndex#of}) and then in SYSTEM.*_PARTITION_STATES we will get UNAVAILABLE state
* instead of the desired one. That's why in {@link #testGlobalPartitionStatesSystemView()} and
* {@link #testLocalPartitionStatesSystemView()} we must manually trigger {@link RaftGroupService#refreshLeader()} that will lead
* partitions to the proper states.
*
- * @param tableName A table whose partitions will do a leader refresh.
+ * @param zoneName A zone whose partitions will do a leader refresh.
* @param partitionsCount Expected the table partitions count for iterating over them.
*/
- private static void waitLeaderOnAllPartitions(String tableName, int partitionsCount) {
+ static void waitLeaderOnAllPartitions(String zoneName, int partitionsCount) {
IgniteImpl node = unwrapIgniteImpl(CLUSTER.node(0));
- TableImpl table = ((PublicApiThreadingTable) node.tables().table(tableName)).unwrap(TableImpl.class);
-
- int tableId = table.tableId();
+ int zoneId = getZoneId(zoneName);
IntStream.range(0, partitionsCount).forEach(partId -> assertThat(
node.replicaManager()
- .replica(new TablePartitionId(tableId, partId))
+ .replica(new ZonePartitionId(zoneId, partId))
.thenCompose(replica -> replica.raftClient().refreshLeader()),
willCompleteSuccessfully()
));
}
- private static String globalPartitionStatesSystemViewSql() {
- return "SELECT ZONE_NAME, TABLE_ID, SCHEMA_NAME, TABLE_NAME, PARTITION_ID, STATE FROM SYSTEM.GLOBAL_PARTITION_STATES";
+ private static String globalZonePartitionStatesSystemViewSql() {
+ return "SELECT ZONE_NAME, PARTITION_ID, PARTITION_STATE, ZONE_ID FROM SYSTEM.GLOBAL_ZONE_PARTITION_STATES "
+ + "WHERE ZONE_NAME != 'Default'";
}
- private static String localPartitionStatesSystemViewSql() {
- return "SELECT NODE_NAME, ZONE_NAME, TABLE_ID, SCHEMA_NAME, TABLE_NAME, PARTITION_ID, STATE, ESTIMATED_ROWS"
- + " FROM SYSTEM.LOCAL_PARTITION_STATES";
+ private static String localZonePartitionStatesSystemViewSql() {
+ return "SELECT NODE_NAME, ZONE_NAME, PARTITION_ID, PARTITION_STATE, ESTIMATED_ROWS, ZONE_ID"
+ + " FROM SYSTEM.LOCAL_ZONE_PARTITION_STATES WHERE ZONE_NAME != 'Default'";
}
- private static int getTableId(String schemaName, String tableName) {
+ private static int getZoneId(String zoneName) {
CatalogManager catalogManager = unwrapIgniteImpl(CLUSTER.aliveNode()).catalogManager();
- return catalogManager.catalog(catalogManager.latestCatalogVersion()).table(schemaName, tableName).id();
+ return catalogManager.catalog(catalogManager.latestCatalogVersion()).zone(zoneName).id();
}
- private static long estimatedSize(String nodeName, String tableName, int partitionId) {
- return CLUSTER.runningNodes()
+ static long estimatedSize(String nodeName, String tableName, int partitionId, Cluster cluster) {
+ return cluster.runningNodes()
.filter(ignite -> nodeName.equals(ignite.name()))
.map(ignite -> {
TableImpl table = unwrapTableImpl(ignite.tables().table(tableName));