You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by zs...@apache.org on 2026/02/16 06:15:17 UTC
(ignite) branch ignite-2.18 updated (2af0e4bf87b -> 0a126f8d717)
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a change to branch ignite-2.18
in repository https://gitbox.apache.org/repos/asf/ignite.git
from 2af0e4bf87b IGNITE-27684 Update IGNITE_HEADER for discovery protocol (#12671)
new a0a665ddc04 IGNITE-27687 Add metrics for inserted/removed bytes to/from cache group - Fixes #12670.
new fba9935f585 IGNITE-27696 Fix IndexQuery can query backup partition if setPartition is used (#12679)
new 0a126f8d717 IGNITE-27457 Add output for enable/disable WAL commands (#12711)
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/_docs/monitoring-metrics/new-metrics.adoc | 2 +
docs/_docs/monitoring-metrics/system-views.adoc | 15 +++
.../ignite/jdbc/thin/JdbcThinMetadataSelfTest.java | 2 +
.../ignite/util/GridCommandHandlerWalTest.java | 80 ++++++++++++--
.../internal/management/wal/WalDisableCommand.java | 8 +-
.../internal/management/wal/WalEnableCommand.java | 8 +-
.../internal/management/wal/WalSetStateTask.java | 52 +++++----
.../management/wal/WalSetStateTaskResult.java | 81 ++++++++++++++
.../ignite/internal/metric/IoStatisticsHolder.java | 14 +++
.../internal/metric/IoStatisticsHolderCache.java | 28 ++++-
.../internal/metric/IoStatisticsHolderIndex.java | 10 ++
.../internal/metric/IoStatisticsHolderNoOp.java | 12 ++
.../internal/metric/IoStatisticsHolderQuery.java | 10 ++
.../persistence/freelist/AbstractFreeList.java | 29 +++--
.../processors/cache/query/CacheQuery.java | 23 ++--
.../query/GridCacheDistributedQueryFuture.java | 17 ++-
.../cache/query/GridCacheQueryManager.java | 6 +-
.../spi/systemview/view/CacheGroupIoView.java | 22 ++++
...IoStatisticsMetricsLocalMXBeanImplSelfTest.java | 122 ++++++++++++++++++++-
.../internal/metric/IoStatisticsSelfTest.java | 2 +-
.../ignite/internal/metric/SystemViewSelfTest.java | 1 +
.../rebalancing/MultiDcRebalancingTest.java | 22 +---
.../junits/common/GridCommonAbstractTest.java | 16 +++
.../cache/query/IndexQueryPartitionTest.java | 5 +-
24 files changed, 495 insertions(+), 92 deletions(-)
create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTaskResult.java
(ignite) 03/03: IGNITE-27457 Add output for enable/disable WAL commands (#12711)
Posted by zs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a commit to branch ignite-2.18
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit 0a126f8d717c65491b821798d14f0ed999ce28c3
Author: Maksim Timonin <ti...@apache.org>
AuthorDate: Thu Feb 12 11:46:54 2026 +0300
IGNITE-27457 Add output for enable/disable WAL commands (#12711)
(cherry picked from commit 38ab8cf87368ae3632f2c177b4b37196bc2a919f)
---
.../ignite/util/GridCommandHandlerWalTest.java | 80 +++++++++++++++++----
.../internal/management/wal/WalDisableCommand.java | 8 ++-
.../internal/management/wal/WalEnableCommand.java | 8 ++-
.../internal/management/wal/WalSetStateTask.java | 52 ++++++++------
.../management/wal/WalSetStateTaskResult.java | 81 ++++++++++++++++++++++
5 files changed, 195 insertions(+), 34 deletions(-)
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerWalTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerWalTest.java
index 4417847b18d..615d0bde022 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerWalTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerWalTest.java
@@ -184,22 +184,78 @@ public class GridCommandHandlerWalTest extends GridCommandHandlerAbstractTest {
srv.cluster().state(ClusterState.ACTIVE);
srv.createCache(new CacheConfiguration<>("cache1")
- .setGroupName("testGroup"));
+ .setGroupName("group1"));
srv.createCache(new CacheConfiguration<>("cache2")
- .setGroupName("testGroup"));
+ .setGroupName("group1"));
+ srv.createCache(new CacheConfiguration<>("cache3")
+ .setGroupName("group2"));
+ srv.createCache("cache4");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "group1,group2,cache4"));
+ outputContains(".*group1.*true.*true.*true.*true.*false");
+ outputContains(".*group2.*true.*true.*true.*true.*false");
+ outputContains(".*cache4.*true.*true.*true.*true.*false");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "disable", "--groups", "group1"));
+ outputContains("Successfully disabled WAL for groups:");
+ outputContains("group1");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "group1"));
+ outputContains(".*group1.*true.*false.*true.*true.*false");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "disable", "--groups", "group1,group2"));
+ outputContains("Successfully disabled WAL for groups:");
+ outputContains("group1");
+ outputContains("group2");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "group1,group2"));
+ outputContains(".*group1.*true.*false.*true.*true.*false");
+ outputContains(".*group2.*true.*false.*true.*true.*false");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "disable", "--groups", "cache4,nonExistentGroup"));
+ outputContains("Successfully disabled WAL for groups:");
+ outputContains("cache4");
+ outputContains("Failed to disable WAL for groups:");
+ outputContains("nonExistentGroup - Cache group not found");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "cache4"));
+ outputContains(".*cache4.*true.*false.*true.*true.*false");
+
+ // Error when using cache name instead of group name.
+ assertEquals(EXIT_CODE_OK, execute("--wal", "enable", "--groups", "cache3"));
+ outputContains("Failed to enable WAL for groups:");
+ outputContains("cache3 - Cache group not found");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "enable", "--groups", "group2,cache4"));
+ outputContains("Successfully enabled WAL for groups:");
+ outputContains("group2");
+ outputContains("cache4");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "group2,cache4"));
+ outputContains(".*group2.*true.*true.*true.*true.*false");
+ outputContains(".*cache4.*true.*true.*true.*true.*false");
+
+ assertEquals(EXIT_CODE_OK, execute("--wal", "disable"));
+ outputContains("Successfully disabled WAL for groups:");
+ outputContains("group1");
+ outputContains("group2");
+ outputContains("cache4");
- assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "testGroup"));
- outputContains(".*testGroup.*true.*true.*true.*true.*false");
-
- assertEquals(EXIT_CODE_OK, execute("--wal", "disable", "--groups", "testGroup"));
-
- assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "testGroup"));
- outputContains(".*testGroup.*true.*false.*true.*true.*false");
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state"));
+ outputContains(".*group1.*true.*false.*true.*true.*false");
+ outputContains(".*group2.*true.*false.*true.*true.*false");
+ outputContains(".*cache4.*true.*false.*true.*true.*false");
- assertEquals(EXIT_CODE_OK, execute("--wal", "enable", "--groups", "testGroup"));
+ assertEquals(EXIT_CODE_OK, execute("--wal", "enable"));
+ outputContains("Successfully enabled WAL for groups:");
+ outputContains("group1");
+ outputContains("group2");
+ outputContains("cache4");
- assertEquals(EXIT_CODE_OK, execute("--wal", "state", "--groups", "testGroup"));
- outputContains(".*testGroup.*true.*true.*true.*true.*false");
+ assertEquals(EXIT_CODE_OK, execute("--wal", "state"));
+ outputContains(".*group1.*true.*true.*true.*true.*false");
+ outputContains(".*group2.*true.*true.*true.*true.*false");
+ outputContains(".*cache4.*true.*true.*true.*true.*false");
}
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalDisableCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalDisableCommand.java
index 9802dc31864..3e3208d93bb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalDisableCommand.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalDisableCommand.java
@@ -17,10 +17,11 @@
package org.apache.ignite.internal.management.wal;
+import java.util.function.Consumer;
import org.apache.ignite.internal.management.api.ComputeCommand;
/** */
-public class WalDisableCommand implements ComputeCommand<WalDisableCommand.WalDisableCommandArg, Void> {
+public class WalDisableCommand implements ComputeCommand<WalDisableCommand.WalDisableCommandArg, WalSetStateTaskResult> {
/** {@inheritDoc} */
@Override public Class<WalSetStateTask> taskClass() {
return WalSetStateTask.class;
@@ -41,6 +42,11 @@ public class WalDisableCommand implements ComputeCommand<WalDisableCommand.WalDi
return "Are you sure? Any node failure without WAL can lead to the loss of all PDS data. CDC events will be lost without WAL.";
}
+ /** {@inheritDoc} */
+ @Override public void printResult(WalDisableCommandArg arg, WalSetStateTaskResult res, Consumer<String> printer) {
+ res.print(false, printer);
+ }
+
/** */
public static class WalDisableCommandArg extends WalStateCommandArg {
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalEnableCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalEnableCommand.java
index 950f416e54a..96f8284c72e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalEnableCommand.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalEnableCommand.java
@@ -17,11 +17,12 @@
package org.apache.ignite.internal.management.wal;
+import java.util.function.Consumer;
import org.apache.ignite.internal.management.api.ComputeCommand;
import org.apache.ignite.internal.management.wal.WalDisableCommand.WalDisableCommandArg;
/** */
-public class WalEnableCommand implements ComputeCommand<WalDisableCommandArg, Void> {
+public class WalEnableCommand implements ComputeCommand<WalDisableCommandArg, WalSetStateTaskResult> {
/** {@inheritDoc} */
@Override public Class<WalSetStateTask> taskClass() {
return WalSetStateTask.class;
@@ -37,6 +38,11 @@ public class WalEnableCommand implements ComputeCommand<WalDisableCommandArg, Vo
return WalEnableCommandArg.class;
}
+ /** {@inheritDoc} */
+ @Override public void printResult(WalDisableCommandArg arg, WalSetStateTaskResult res, Consumer<String> printer) {
+ res.print(true, printer);
+ }
+
/** */
public static class WalEnableCommandArg extends WalDisableCommandArg {
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTask.java
index 017dd729bb5..af71cb07e78 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTask.java
@@ -17,37 +17,34 @@
package org.apache.ignite.internal.management.wal;
+import java.util.ArrayList;
import java.util.Arrays;
-import java.util.HashSet;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.ignite.IgniteException;
-import org.apache.ignite.compute.ComputeJobResult;
import org.apache.ignite.internal.management.wal.WalDisableCommand.WalDisableCommandArg;
import org.apache.ignite.internal.management.wal.WalEnableCommand.WalEnableCommandArg;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.visor.VisorJob;
-import org.apache.ignite.internal.visor.VisorMultiNodeTask;
+import org.apache.ignite.internal.visor.VisorOneNodeTask;
import org.jetbrains.annotations.Nullable;
/** */
-public class WalSetStateTask extends VisorMultiNodeTask<WalDisableCommandArg, Void, Void> {
+public class WalSetStateTask extends VisorOneNodeTask<WalDisableCommandArg, WalSetStateTaskResult> {
/** */
private static final long serialVersionUID = 0;
/** {@inheritDoc} */
- @Override protected VisorJob<WalDisableCommandArg, Void> job(WalDisableCommandArg arg) {
- return new WalDisableJob(arg, false);
- }
-
- /** {@inheritDoc} */
- @Override protected @Nullable Void reduce0(List<ComputeJobResult> res) throws IgniteException {
- return null;
+ @Override protected VisorJob<WalDisableCommandArg, WalSetStateTaskResult> job(WalDisableCommandArg arg) {
+ return new WalDisableJob(arg, debug);
}
/** */
- private static class WalDisableJob extends VisorJob<WalDisableCommandArg, Void> {
+ private static class WalDisableJob extends VisorJob<WalDisableCommandArg, WalSetStateTaskResult> {
/** */
private static final long serialVersionUID = 0;
@@ -57,22 +54,37 @@ public class WalSetStateTask extends VisorMultiNodeTask<WalDisableCommandArg, Vo
}
/** {@inheritDoc} */
- @Override protected Void run(@Nullable WalDisableCommandArg arg) throws IgniteException {
- Set<String> grps = F.isEmpty(arg.groups()) ? null : new HashSet<>(Arrays.asList(arg.groups()));
+ @Override protected WalSetStateTaskResult run(@Nullable WalDisableCommandArg arg) throws IgniteException {
+ Set<String> grps = arg.groups == null ? null : Arrays.stream(arg.groups).collect(Collectors.toSet());
+
+ List<String> successGrps = new ArrayList<>();
+ Map<String, String> failedGrps = new HashMap<>();
for (CacheGroupContext gctx : ignite.context().cache().cacheGroups()) {
String grpName = gctx.cacheOrGroupName();
- if (grps != null && !grps.contains(grpName))
+ if (grps != null && !grps.remove(grpName))
continue;
- if (arg instanceof WalEnableCommandArg)
- ignite.cluster().enableWal(grpName);
- else
- ignite.cluster().disableWal(grpName);
+ try {
+ if (arg instanceof WalEnableCommandArg)
+ ignite.cluster().enableWal(grpName);
+ else
+ ignite.cluster().disableWal(grpName);
+
+ successGrps.add(grpName);
+ }
+ catch (IgniteException e) {
+ failedGrps.put(grpName, e.getMessage());
+ }
+ }
+
+ if (!F.isEmpty(grps)) {
+ for (String grp: grps)
+ failedGrps.put(grp, "Cache group not found");
}
- return null;
+ return new WalSetStateTaskResult(successGrps, failedGrps);
}
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTaskResult.java
new file mode 100644
index 00000000000..53e0f8baad9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/management/wal/WalSetStateTaskResult.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.management.wal;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.dto.IgniteDataTransferObject;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ * Result of WAL enable/disable operation.
+ */
+public class WalSetStateTaskResult extends IgniteDataTransferObject {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Successfully processed groups. */
+ @Order(0)
+ List<String> successGrps;
+
+ /** Errors by group name. */
+ @Order(1)
+ Map<String, String> errorsByGrp;
+
+ /** Default constructor. */
+ public WalSetStateTaskResult() {
+ // No-op.
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param successGrps Successfully processed groups.
+ * @param errorsByGrp Error messages.
+ */
+ public WalSetStateTaskResult(List<String> successGrps, Map<String, String> errorsByGrp) {
+ this.successGrps = successGrps;
+ this.errorsByGrp = errorsByGrp;
+ }
+
+ /**
+ * Print WAL disable/enable command result.
+ *
+ * @param enable If {@code true} then "enable" operation, otherwise "disable".
+ * @param printer Output consumer.
+ */
+ void print(boolean enable, Consumer<String> printer) {
+ String op = enable ? "enable" : "disable";
+
+ if (!successGrps.isEmpty()) {
+ printer.accept("Successfully " + op + "d WAL for groups:");
+
+ for (String grp : successGrps)
+ printer.accept(" " + grp);
+ }
+
+ if (!F.isEmpty(errorsByGrp)) {
+ printer.accept("Failed to " + op + " WAL for groups:");
+
+ for (Map.Entry<String, String> entry : errorsByGrp.entrySet())
+ printer.accept(" " + entry.getKey() + " - " + entry.getValue());
+ }
+ }
+}
(ignite) 01/03: IGNITE-27687 Add metrics for inserted/removed bytes to/from cache group - Fixes #12670.
Posted by zs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a commit to branch ignite-2.18
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit a0a665ddc047d201601cf2db7b541e9a1c71b5cf
Author: Aleksey Plekhanov <pl...@gmail.com>
AuthorDate: Fri Feb 13 17:54:54 2026 +0300
IGNITE-27687 Add metrics for inserted/removed bytes to/from cache group - Fixes #12670.
Signed-off-by: Aleksey Plekhanov <pl...@gmail.com>
(cherry picked from commit 1154765f9d13b316c769246eed26e11930408344)
---
docs/_docs/monitoring-metrics/new-metrics.adoc | 2 +
docs/_docs/monitoring-metrics/system-views.adoc | 15 +++
.../ignite/jdbc/thin/JdbcThinMetadataSelfTest.java | 2 +
.../ignite/internal/metric/IoStatisticsHolder.java | 14 +++
.../internal/metric/IoStatisticsHolderCache.java | 28 ++++-
.../internal/metric/IoStatisticsHolderIndex.java | 10 ++
.../internal/metric/IoStatisticsHolderNoOp.java | 12 ++
.../internal/metric/IoStatisticsHolderQuery.java | 10 ++
.../persistence/freelist/AbstractFreeList.java | 29 +++--
.../spi/systemview/view/CacheGroupIoView.java | 22 ++++
...IoStatisticsMetricsLocalMXBeanImplSelfTest.java | 122 ++++++++++++++++++++-
.../internal/metric/IoStatisticsSelfTest.java | 2 +-
.../ignite/internal/metric/SystemViewSelfTest.java | 1 +
.../rebalancing/MultiDcRebalancingTest.java | 22 +---
.../junits/common/GridCommonAbstractTest.java | 16 +++
15 files changed, 272 insertions(+), 35 deletions(-)
diff --git a/docs/_docs/monitoring-metrics/new-metrics.adoc b/docs/_docs/monitoring-metrics/new-metrics.adoc
index 862f27a3391..19a2ca3121e 100644
--- a/docs/_docs/monitoring-metrics/new-metrics.adoc
+++ b/docs/_docs/monitoring-metrics/new-metrics.adoc
@@ -244,6 +244,8 @@ Register name: `io.statistics.cacheGroups.{group_name}`
|grpId | integer | Group id
|name | string | Name of the index
|startTime | long | Statistics collect start time
+|insertedBytes | long | Count of inserted to store bytes
+|removedBytes | long | Count of removed from store bytes
|===
diff --git a/docs/_docs/monitoring-metrics/system-views.adoc b/docs/_docs/monitoring-metrics/system-views.adoc
index f9afb0ccc90..ad4589bb23c 100644
--- a/docs/_docs/monitoring-metrics/system-views.adoc
+++ b/docs/_docs/monitoring-metrics/system-views.adoc
@@ -789,6 +789,21 @@ This view exposes information about the distribution of cache group partitions a
|IS_PRIMARY | boolean | Primary partition flag
|===
+== LOCAL_CACHE_GROUPS_IO
+
+This view exposes information about local node IO statistics for cache groups.
+
+[{table_opts}]
+|===
+|Column | Data type | Description
+|CACHE_GROUP_ID | int| Cache group ID
+|CACHE_GROUP_NAME | string| Cache group name
+|PHYSICAL_READS | long | Count of logical page reads
+|LOGICAL_READS | long | Count of physical page reads
+|INSERTED_BYTES | long | Count of inserted to store bytes
+|REMOVED_BYTES | long | Count of removed from store bytes
+|===
+
== BINARY_METADATA
This view exposes information about all available binary types.
diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
index bebf1e48861..ebcd723f83c 100644
--- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java
@@ -767,6 +767,8 @@ public class JdbcThinMetadataSelfTest extends JdbcThinAbstractSelfTest {
"SYS.LOCAL_CACHE_GROUPS_IO.CACHE_GROUP_NAME.null",
"SYS.LOCAL_CACHE_GROUPS_IO.PHYSICAL_READS.null",
"SYS.LOCAL_CACHE_GROUPS_IO.LOGICAL_READS.null",
+ "SYS.LOCAL_CACHE_GROUPS_IO.INSERTED_BYTES.null",
+ "SYS.LOCAL_CACHE_GROUPS_IO.REMOVED_BYTES.null",
"SYS.SQL_QUERIES_HISTORY.SCHEMA_NAME.null",
"SYS.SQL_QUERIES_HISTORY.SQL.null",
"SYS.SQL_QUERIES_HISTORY.LOCAL.null",
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolder.java b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolder.java
index 28604d063b0..f4f2afd2aa7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolder.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolder.java
@@ -36,6 +36,20 @@ public interface IoStatisticsHolder {
*/
public void trackPhysicalAndLogicalRead(long pageAddr);
+ /**
+ * Track insert data to the page.
+ *
+ * @param bytes Bytes inserted.
+ */
+ public void trackPageInsertData(long bytes);
+
+ /**
+ * Track remove data from the page.
+ *
+ * @param bytes Bytes removed.
+ */
+ public void trackPageRemoveData(long bytes);
+
/**
* @return Number of logical reads.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderCache.java b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderCache.java
index 11e3330fc42..c8d290c656f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderCache.java
@@ -38,12 +38,24 @@ public class IoStatisticsHolderCache implements IoStatisticsHolder {
/** */
public static final String LOGICAL_READS = "LOGICAL_READS";
+ /** */
+ public static final String INSERTED_BYTES = "insertedBytes";
+
+ /** */
+ public static final String REMOVED_BYTES = "removedBytes";
+
/** */
private final LongAdderMetric logicalReadCtr;
/** */
private final LongAdderMetric physicalReadCtr;
+ /** */
+ private final LongAdderMetric insertedBytes;
+
+ /** */
+ private final LongAdderMetric removedBytes;
+
/** */
private final String grpName;
@@ -67,8 +79,10 @@ public class IoStatisticsHolderCache implements IoStatisticsHolder {
mreg.objectMetric("name", String.class, null).value(grpName);
mreg.intMetric("grpId", null).value(grpId);
- this.logicalReadCtr = mreg.longAdderMetric(LOGICAL_READS, null);
- this.physicalReadCtr = mreg.longAdderMetric(PHYSICAL_READS, null);
+ logicalReadCtr = mreg.longAdderMetric(LOGICAL_READS, "Count of logical page reads");
+ physicalReadCtr = mreg.longAdderMetric(PHYSICAL_READS, "Count of physical page reads");
+ insertedBytes = mreg.longAdderMetric(INSERTED_BYTES, "Count of inserted to store bytes");
+ removedBytes = mreg.longAdderMetric(REMOVED_BYTES, "Count of removed from store bytes");
}
/** {@inheritDoc} */
@@ -95,6 +109,16 @@ public class IoStatisticsHolderCache implements IoStatisticsHolder {
}
}
+ /** {@inheritDoc} */
+ @Override public void trackPageInsertData(long bytes) {
+ insertedBytes.add(bytes);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void trackPageRemoveData(long bytes) {
+ removedBytes.add(bytes);
+ }
+
/** {@inheritDoc} */
@Override public long logicalReads() {
return logicalReadCtr.value();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderIndex.java b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderIndex.java
index 8cbea95b412..1bce5e4add7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderIndex.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderIndex.java
@@ -141,6 +141,16 @@ public class IoStatisticsHolderIndex implements IoStatisticsHolder {
}
}
+ /** {@inheritDoc} */
+ @Override public void trackPageInsertData(long bytes) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void trackPageRemoveData(long bytes) {
+ // No-op.
+ }
+
/** {@inheritDoc} */
@Override public long logicalReads() {
return logicalReadLeafCtr.value() + logicalReadInnerCtr.value();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderNoOp.java b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderNoOp.java
index 936fb11b5db..fe5c1c95907 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderNoOp.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderNoOp.java
@@ -34,10 +34,22 @@ public class IoStatisticsHolderNoOp implements IoStatisticsHolder {
/** {@inheritDoc} */
@Override public void trackLogicalRead(long pageAddr) {
+ // No-op.
}
/** {@inheritDoc} */
@Override public void trackPhysicalAndLogicalRead(long pageAddr) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void trackPageInsertData(long bytes) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void trackPageRemoveData(long bytes) {
+ // No-op.
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderQuery.java b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderQuery.java
index b7884e521de..232c39a8db2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/metric/IoStatisticsHolderQuery.java
@@ -53,6 +53,16 @@ public class IoStatisticsHolderQuery implements IoStatisticsHolder {
physicalReadCtr.increment();
}
+ /** {@inheritDoc} */
+ @Override public void trackPageInsertData(long bytes) {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void trackPageRemoveData(long bytes) {
+ // No-op.
+ }
+
/** {@inheritDoc} */
@Override public long logicalReads() {
return logicalReadCtr.longValue();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
index 8165389599f..62452095631 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/freelist/AbstractFreeList.java
@@ -135,6 +135,9 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
PageUtils.getBytes(pageAddr, data.offset(), payload, 0, rowSize);
+ statHolder.trackPageRemoveData(rowSize);
+ statHolder.trackPageInsertData(rowSize);
+
wal.log(new DataPageUpdateRecord(
cacheId,
pageId,
@@ -164,9 +167,9 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
Boolean walPlc,
T row,
int written,
- IoStatisticsHolder statHolder)
- throws IgniteCheckedException {
- written = addRow(pageId, page, pageAddr, iox, row, written);
+ IoStatisticsHolder statHolder
+ ) throws IgniteCheckedException {
+ written = addRow(pageId, page, pageAddr, iox, row, written, statHolder);
putPage(((AbstractDataPageIO)iox).getFreeSpace(pageAddr), pageId, page, pageAddr, statHolder);
@@ -180,6 +183,7 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
* @param iox IO.
* @param row Row to write.
* @param written Written size.
+ * @param statHolder IO statistics holder.
* @return Number of bytes written, {@link #COMPLETE} if the row was fully written.
* @throws IgniteCheckedException If failed.
*/
@@ -189,8 +193,9 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
long pageAddr,
PageIO iox,
T row,
- int written)
- throws IgniteCheckedException {
+ int written,
+ IoStatisticsHolder statHolder
+ ) throws IgniteCheckedException {
AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
long lastLink = row.link();
@@ -209,6 +214,8 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
evictionTracker.trackFragmentPage(pageId, lastLink, written == rowSize);
+ statHolder.trackPageInsertData(oldFreeSpace - io.getFreeSpace(pageAddr));
+
// Avoid boxing with garbage generation for usual case.
return written == rowSize ? COMPLETE : written;
}
@@ -324,8 +331,8 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
Boolean walPlc,
GridCursor<T> cur,
int written,
- IoStatisticsHolder statHolder)
- throws IgniteCheckedException {
+ IoStatisticsHolder statHolder
+ ) throws IgniteCheckedException {
AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
// Fill the page up to the end.
@@ -341,7 +348,7 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
break;
}
- written = writeRowHnd.addRow(pageId, page, pageAddr, io, row, written);
+ written = writeRowHnd.addRow(pageId, page, pageAddr, io, row, written, statHolder);
assert written == COMPLETE;
}
@@ -377,8 +384,8 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
Boolean walPlc,
ReuseBag reuseBag,
int itemId,
- IoStatisticsHolder statHolder)
- throws IgniteCheckedException {
+ IoStatisticsHolder statHolder
+ ) throws IgniteCheckedException {
AbstractDataPageIO<T> io = (AbstractDataPageIO<T>)iox;
int oldFreeSpace = io.getFreeSpace(pageAddr);
@@ -418,6 +425,8 @@ public abstract class AbstractFreeList<T extends Storable> extends PagesList imp
put(null, pageId, page, pageAddr, newBucket, statHolder);
}
+ statHolder.trackPageRemoveData(newFreeSpace - oldFreeSpace);
+
// For common case boxed 0L will be cached inside of Long, so no garbage will be produced.
return nextLink;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/CacheGroupIoView.java b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/CacheGroupIoView.java
index 328d956bb1e..46e07d11934 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/CacheGroupIoView.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/CacheGroupIoView.java
@@ -22,8 +22,10 @@ import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.metric.MetricRegistry;
import org.apache.ignite.spi.metric.LongMetric;
+import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.INSERTED_BYTES;
import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.LOGICAL_READS;
import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.PHYSICAL_READS;
+import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.REMOVED_BYTES;
/**
* Cache group IO statistics representation for a {@link SystemView}.
@@ -79,4 +81,24 @@ public class CacheGroupIoView {
return metric != null ? metric.value() : 0;
}
+
+ /**
+ * @return Inserted bytes into store.
+ */
+ @Order(4)
+ public long insertedBytes() {
+ LongMetric metric = mreg.findMetric(INSERTED_BYTES);
+
+ return metric != null ? metric.value() : 0;
+ }
+
+ /**
+ * @return Removed bytes from store.
+ */
+ @Order(5)
+ public long removedBytes() {
+ LongMetric metric = mreg.findMetric(REMOVED_BYTES);
+
+ return metric != null ? metric.value() : 0;
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsMetricsLocalMXBeanImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsMetricsLocalMXBeanImplSelfTest.java
index ec5d77d3781..cf53734917f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsMetricsLocalMXBeanImplSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsMetricsLocalMXBeanImplSelfTest.java
@@ -27,9 +27,12 @@ import org.apache.ignite.internal.processors.metric.GridMetricManager;
import org.apache.ignite.metric.MetricRegistry;
import org.apache.ignite.spi.metric.LongMetric;
import org.apache.ignite.spi.metric.ReadOnlyMetricRegistry;
+import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
+import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.INSERTED_BYTES;
+import static org.apache.ignite.internal.metric.IoStatisticsHolderCache.REMOVED_BYTES;
import static org.apache.ignite.internal.metric.IoStatisticsHolderIndex.HASH_PK_IDX_NAME;
import static org.apache.ignite.internal.metric.IoStatisticsHolderIndex.LOGICAL_READS_INNER;
import static org.apache.ignite.internal.metric.IoStatisticsHolderIndex.LOGICAL_READS_LEAF;
@@ -62,12 +65,15 @@ public class IoStatisticsMetricsLocalMXBeanImplSelfTest extends GridCommonAbstra
}
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- super.beforeTestsStarted();
-
+ @Override protected void beforeTest() throws Exception {
ignite = startGrid(0);
}
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
/**
* Simple test JMX bean for indexes IO stats.
*
@@ -130,6 +136,116 @@ public class IoStatisticsMetricsLocalMXBeanImplSelfTest extends GridCommonAbstra
assertEquals(0, cachePhysicalReadsCnt);
}
+ /** */
+ @Test
+ public void testInsertedDeletedBytes() {
+ int cnt = 100;
+
+ MetricRegistry mreg = ignite.context().metric()
+ .registry(metricName(CACHE_GROUP.metricGroupName(), DEFAULT_CACHE_NAME));
+
+ LongMetric insertedBytes = mreg.findMetric(INSERTED_BYTES);
+ LongMetric removedBytes = mreg.findMetric(REMOVED_BYTES);
+
+ assertEquals(0, insertedBytes.value());
+
+ populateCache(cnt);
+
+ int minEntrySize = 20; // Size of key, size of val, entry headers, data page payload headers, etc.
+ int maxEntrySize = 100;
+
+ assertTrue(insertedBytes.value() > cnt * minEntrySize);
+ assertTrue(insertedBytes.value() < cnt * maxEntrySize);
+
+ assertEquals(0, removedBytes.value());
+
+ clearCache(cnt);
+
+ assertEquals(insertedBytes.value(), removedBytes.value());
+ }
+
+ /** */
+ @Test
+ public void testInsertedDeletedBytesInplaceUpdate() {
+ MetricRegistry mreg = ignite.context().metric()
+ .registry(metricName(CACHE_GROUP.metricGroupName(), DEFAULT_CACHE_NAME));
+
+ LongMetric insertedBytes = mreg.findMetric(INSERTED_BYTES);
+ LongMetric removedBytes = mreg.findMetric(REMOVED_BYTES);
+
+ ignite.cache(DEFAULT_CACHE_NAME).put(0, 0);
+
+ long inserted0 = insertedBytes.value();
+
+ assertNotSame(0, inserted0);
+
+ // Inplace update.
+ ignite.cache(DEFAULT_CACHE_NAME).put(0, 1);
+
+ assertNotSame(0, removedBytes.value());
+ assertEquals(insertedBytes.value() - inserted0, removedBytes.value());
+ }
+
+ /** */
+ @Test
+ public void testInsertedDeletedBytesMultipage() {
+ MetricRegistry mreg = ignite.context().metric()
+ .registry(metricName(CACHE_GROUP.metricGroupName(), DEFAULT_CACHE_NAME));
+
+ LongMetric insertedBytes = mreg.findMetric(INSERTED_BYTES);
+ LongMetric removedBytes = mreg.findMetric(REMOVED_BYTES);
+
+ assertEquals(0, insertedBytes.value());
+
+ int size = 100_000;
+
+ ignite.cache(DEFAULT_CACHE_NAME).put(0, new byte[size]);
+
+ assertTrue("Unexpected value for insertedBytes: " + insertedBytes.value(), insertedBytes.value() > size);
+
+ ignite.cache(DEFAULT_CACHE_NAME).remove(0);
+
+ assertEquals(insertedBytes.value(), removedBytes.value());
+ }
+
+ /** */
+ @Test
+ public void testInsertedDeletedBytesOnRebalance() throws Exception {
+ int cnt = 100;
+
+ MetricRegistry mreg0 = ignite.context().metric()
+ .registry(metricName(CACHE_GROUP.metricGroupName(), DEFAULT_CACHE_NAME));
+
+ LongMetric insertedBytes0 = mreg0.findMetric(INSERTED_BYTES);
+ LongMetric removedBytes0 = mreg0.findMetric(REMOVED_BYTES);
+
+ populateCache(cnt);
+
+ assertNotSame(0, insertedBytes0.value());
+ assertEquals(0, removedBytes0.value());
+
+ IgniteEx ignite1 = startGrid(1);
+
+ waitRebalanceFinished(ignite1, DEFAULT_CACHE_NAME);
+
+ MetricRegistry mreg1 = ignite1.context().metric()
+ .registry(metricName(CACHE_GROUP.metricGroupName(), DEFAULT_CACHE_NAME));
+
+ LongMetric insertedBytes1 = mreg1.findMetric(INSERTED_BYTES);
+ LongMetric removedBytes1 = mreg1.findMetric(REMOVED_BYTES);
+
+ assertNotSame(0, removedBytes0.value());
+ assertNotSame(0, insertedBytes1.value());
+ assertEquals(0, removedBytes1.value());
+
+ clearCache(cnt);
+
+ assertNotSame(0, removedBytes1.value());
+
+ assertTrue(GridTestUtils.waitForCondition(() -> insertedBytes0.value() == removedBytes0.value(), 1_000L));
+ assertTrue(GridTestUtils.waitForCondition(() -> insertedBytes1.value() == removedBytes1.value(), 1_000L));
+ }
+
/**
* @param cnt Number of inserting elements.
*/
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsSelfTest.java
index 980fe203aae..6eca1644f62 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/metric/IoStatisticsSelfTest.java
@@ -98,7 +98,7 @@ public class IoStatisticsSelfTest extends GridCommonAbstractTest {
assertNotNull(mreg);
if (type == CACHE_GROUP) {
- assertEquals(5, Iterators.size(mreg.iterator()));
+ assertEquals(7, Iterators.size(mreg.iterator()));
assertEquals(0, mreg.<LongMetric>findMetric(LOGICAL_READS).value());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewSelfTest.java
index b27fdbd938c..e1694888690 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/metric/SystemViewSelfTest.java
@@ -940,6 +940,7 @@ public class SystemViewSelfTest extends GridCommonAbstractTest {
assertNotNull(row);
assertTrue(row.logicalReads() > 0);
+ assertTrue(row.insertedBytes() > 0);
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/MultiDcRebalancingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/MultiDcRebalancingTest.java
index bb71087a7e7..9f77dc49626 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/MultiDcRebalancingTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/rebalancing/MultiDcRebalancingTest.java
@@ -27,16 +27,13 @@ import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
-import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemander;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
-import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.WithSystemProperty;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -122,8 +119,8 @@ public class MultiDcRebalancingTest extends GridCommonAbstractTest {
resetBaselineTopology();
- waitRebalanceFinished(ignite2);
- waitRebalanceFinished(ignite3);
+ waitRebalanceFinished(ignite2, DEFAULT_CACHE_NAME);
+ waitRebalanceFinished(ignite3, DEFAULT_CACHE_NAME);
assertTrue(commSPI(ignite2).rebalanceMsgCnt > 0);
assertTrue(commSPI(ignite3).rebalanceMsgCnt > 0);
@@ -158,7 +155,7 @@ public class MultiDcRebalancingTest extends GridCommonAbstractTest {
IgniteEx ignite2 = startGrid(2, DC1);
- waitRebalanceFinished(ignite2);
+ waitRebalanceFinished(ignite2, DEFAULT_CACHE_NAME);
assertTrue(commSPI(ignite2).rebalanceMsgCnt > 0);
assertTrue(commSPI(ignite2).historical);
@@ -170,19 +167,6 @@ public class MultiDcRebalancingTest extends GridCommonAbstractTest {
return (RebalanceAwareCommSPI)(ignite.configuration().getCommunicationSpi());
}
- /** */
- private void waitRebalanceFinished(IgniteEx ignite) throws Exception {
- assertTrue(GridTestUtils.waitForCondition(() -> {
- IgniteInternalFuture<Boolean> fut = ignite.cachex(DEFAULT_CACHE_NAME).context().preloader().rebalanceFuture();
-
- GridDhtPartitionDemander.RebalanceFuture rebFut = (GridDhtPartitionDemander.RebalanceFuture)fut;
-
- return (!rebFut.isInitial() && rebFut.topologyVersion().topologyVersion() == ignite.cluster().topologyVersion());
- }, 1000));
-
- assertTrue(ignite.cachex(DEFAULT_CACHE_NAME).context().preloader().rebalanceFuture().get());
- }
-
/** */
private static class RebalanceAwareCommSPI extends TcpCommunicationSpi {
/** */
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index d3d9efffaa5..8642fcf8611 100755
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -103,6 +103,7 @@ import org.apache.ignite.internal.processors.cache.WalStateManager.WALDisableCon
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedCache;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemander;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionFullMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMap;
@@ -2858,4 +2859,19 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
protected static Marshaller marshaller(Ignite ign) {
return ((IgniteEx)ign).context().marshaller();
}
+
+ /**
+ * Wait for rebalance on current topology finished.
+ */
+ protected static void waitRebalanceFinished(IgniteEx ignite, String cacheName) throws Exception {
+ assertTrue(GridTestUtils.waitForCondition(() -> {
+ IgniteInternalFuture<Boolean> fut = ignite.cachex(cacheName).context().preloader().rebalanceFuture();
+
+ GridDhtPartitionDemander.RebalanceFuture rebFut = (GridDhtPartitionDemander.RebalanceFuture)fut;
+
+ return (!rebFut.isInitial() && rebFut.topologyVersion().topologyVersion() == ignite.cluster().topologyVersion());
+ }, 1000));
+
+ assertTrue(ignite.cachex(cacheName).context().preloader().rebalanceFuture().get());
+ }
}
(ignite) 02/03: IGNITE-27696 Fix IndexQuery can query backup partition if setPartition is used (#12679)
Posted by zs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zstan pushed a commit to branch ignite-2.18
in repository https://gitbox.apache.org/repos/asf/ignite.git
commit fba9935f5855537850f9280d71577cd18f7f580a
Author: Maksim Timonin <ti...@apache.org>
AuthorDate: Fri Feb 13 15:52:51 2026 +0300
IGNITE-27696 Fix IndexQuery can query backup partition if setPartition is used (#12679)
(cherry picked from commit 4e2e691fa9076e630eba928202d59a18b96bb5fd)
---
.../processors/cache/query/CacheQuery.java | 23 ++++++++++------------
.../query/GridCacheDistributedQueryFuture.java | 17 +++++++++++-----
.../cache/query/GridCacheQueryManager.java | 6 +++---
.../cache/query/IndexQueryPartitionTest.java | 5 +++--
4 files changed, 28 insertions(+), 23 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQuery.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQuery.java
index 6f394c2efce..9c6d6782b1c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/CacheQuery.java
@@ -62,7 +62,6 @@ import org.apache.ignite.internal.util.lang.GridCloseableIterator;
import org.apache.ignite.internal.util.lang.GridIterator;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.P1;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
@@ -840,9 +839,12 @@ public class CacheQuery<T> {
}
/**
+ * Collects query data nodes matching specified {@code prj} and {@code part}.
+ *
* @param cctx Cache context.
* @param prj Projection (optional).
- * @return Collection of data nodes in provided projection (if any).
+ * @param part Partition (optional).
+ * @return Collection of data nodes matching specified {@code prj} and {@code part}.
* @throws IgniteCheckedException If partition number is invalid.
*/
private static Collection<ClusterNode> nodes(final GridCacheContext<?, ?> cctx,
@@ -856,19 +858,14 @@ public class CacheQuery<T> {
if (prj == null && part == null)
return affNodes;
- if (part != null && part >= cctx.affinity().partitions())
- throw new IgniteCheckedException("Invalid partition number: " + part);
+ if (part != null) {
+ if (part >= cctx.affinity().partitions())
+ throw new IgniteCheckedException("Invalid partition number: " + part);
- final Set<ClusterNode> owners =
- part == null ? Collections.<ClusterNode>emptySet() : new HashSet<>(cctx.topology().owners(part, topVer));
+ affNodes = cctx.topology().nodes(part, topVer);
+ }
- return F.view(affNodes, new P1<ClusterNode>() {
- @Override public boolean apply(ClusterNode n) {
- return cctx.discovery().cacheAffinityNode(n, cctx.name()) &&
- (prj == null || prj.node(n.id()) != null) &&
- (part == null || owners.contains(n));
- }
- });
+ return prj == null ? affNodes : F.view(affNodes, n -> prj.node(n.id()) != null);
}
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
index 2b12a3b784c..515cc7323b5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheDistributedQueryFuture.java
@@ -25,6 +25,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.IgniteCheckedException;
@@ -38,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.query.reducer.NodePageStream;
import org.apache.ignite.internal.processors.cache.query.reducer.TextQueryReducer;
import org.apache.ignite.internal.processors.cache.query.reducer.UnsortedCacheQueryReducer;
import org.apache.ignite.internal.util.lang.GridPlainCallable;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.INDEX;
@@ -92,7 +94,7 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
qryMgr = (GridCacheDistributedQueryManager<K, V>)ctx.queries();
if (qry.query().partition() != null)
- nodes = Collections.singletonList(node(nodes));
+ nodes = Collections.singletonList(cctx.isReplicated() ? localOrRemoteNode(nodes) : F.first(nodes));
streams = new ConcurrentHashMap<>(nodes.size());
@@ -118,18 +120,23 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
}
/**
- * @return Nodes for query execution.
+ * @return A local node if available, otherwise a random node from the given collection.
*/
- private ClusterNode node(Collection<ClusterNode> nodes) {
+ private ClusterNode localOrRemoteNode(Collection<ClusterNode> nodes) {
+ int remoteNodeIdx = ThreadLocalRandom.current().nextInt(nodes.size());
+
ClusterNode rmtNode = null;
for (ClusterNode node : nodes) {
if (node.isLocal())
return node;
- rmtNode = node;
+ if (remoteNodeIdx-- == 0)
+ rmtNode = node;
}
+ assert rmtNode != null;
+
return rmtNode;
}
@@ -282,7 +289,7 @@ public class GridCacheDistributedQueryFuture<K, V, R> extends GridCacheQueryFutu
GridCacheQueryRequest req = GridCacheQueryRequest.cancelRequest(cctx, reqId, fields());
if (nodeId.equals(cctx.localNodeId())) {
- // Process cancel query directly (without sending) for local node,
+ // Process cancel query directly (without sending) for local node.
cctx.closures().callLocalSafe(new GridPlainCallable<Object>() {
@Override public Object call() {
qryMgr.processQueryRequest(cctx.localNodeId(), req);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
index 8fe2c925bb1..2db3ed5263e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
@@ -1121,11 +1121,11 @@ public abstract class GridCacheQueryManager<K, V> extends GridCacheManagerAdapte
int[] parts = null;
if (part != null) {
- final GridDhtLocalPartition locPart = cctx.dht().topology().localPartition(part);
+ AffinityTopologyVersion topVer = cctx.affinity().affinityTopologyVersion();
- if (locPart == null || locPart.state() != OWNING) {
+ if (cctx.isPartitioned() && !cctx.affinity().primaryByPartition(cctx.localNode(), part, topVer)) {
throw new CacheInvalidStateException("Failed to execute index query because required partition " +
- "has not been found on local node [cacheName=" + cctx.name() + ", part=" + part + "]");
+ "is not primary on local node [cacheName=" + cctx.name() + ", part=" + part + ", topVer=" + topVer + ']');
}
parts = new int[] {part};
diff --git a/modules/indexing/src/test/java/org/apache/ignite/cache/query/IndexQueryPartitionTest.java b/modules/indexing/src/test/java/org/apache/ignite/cache/query/IndexQueryPartitionTest.java
index 73ea7f30532..8fe205ef7d9 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/cache/query/IndexQueryPartitionTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/cache/query/IndexQueryPartitionTest.java
@@ -78,6 +78,7 @@ public class IndexQueryPartitionTest extends GridCommonAbstractTest {
.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
.setCacheMode(cacheMode)
.setIndexedTypes(Integer.class, Person.class)
+ .setBackups(1)
.setAffinity(new RendezvousAffinityFunction().setPartitions(100));
cfg.setCacheConfiguration(ccfg);
@@ -136,7 +137,7 @@ public class IndexQueryPartitionTest extends GridCommonAbstractTest {
}
}
- assertEquals(sendReq, TestRecordingCommunicationSpi.spi(grid()).recordedMessages(true).size());
+ assertEquals("part=" + part, sendReq, TestRecordingCommunicationSpi.spi(grid()).recordedMessages(true).size());
}
}
@@ -181,7 +182,7 @@ public class IndexQueryPartitionTest extends GridCommonAbstractTest {
GridTestUtils.assertThrows(null, () -> grid().cache("CACHE").query(qry).getAll(),
client ? IgniteException.class : CacheInvalidStateException.class,
client ? "Failed to execute local index query on a client node." :
- "Failed to execute index query because required partition has not been found on local node");
+ "Failed to execute index query because required partition is not primary on local node");
}
else
assertTrue(!grid().cache("CACHE").query(qry).getAll().isEmpty());