IGNITE-18834 Index rebuild ducktest

modules/core/src/main/java/org/apache/ignite/internal/managers/indexing/IndexesRebuildTask.java

@@ -97,7 +97,7 @@ public class IndexesRebuildTask {
 
             if (log.isInfoEnabled()) {
                 log.info("WAL disabled for index partition " +
-                    "[name=" + cctx.group().name() + ", id=" + cctx.groupId() + ']');
+                    "[name=" + cctx.group().cacheOrGroupName() + ", id=" + cctx.groupId() + ']');
             }
         }
 
@@ -151,7 +151,7 @@ public class IndexesRebuildTask {
 
                     if (log.isInfoEnabled()) {
                         log.info("WAL enabled for index partition " +
-                            "[name=" + cctx.group().name() + ", id=" + cctx.group().groupId() + ']');
+                            "[name=" + cctx.group().cacheOrGroupName() + ", id=" + cctx.group().groupId() + ']');
                     }
                 }
             }

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/DataGenerationApplication.java

@@ -17,12 +17,17 @@
 
 package org.apache.ignite.internal.ducktest.tests;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.ThreadLocalRandom;
 import com.fasterxml.jackson.databind.JsonNode;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteDataStreamer;
 import org.apache.ignite.binary.BinaryObject;
 import org.apache.ignite.binary.BinaryObjectBuilder;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.internal.ducktest.utils.IgniteAwareApplication;
 
@@ -33,6 +38,9 @@ public class DataGenerationApplication extends IgniteAwareApplication {
     /** Max streamer data size. */
     private static final int MAX_STREAMER_DATA_SIZE = 100_000_000;
 
+    /** */
+    public static final String VAL_TYPE = "org.apache.ignite.ducktest.DataBinary";
+
     /** {@inheritDoc} */
     @Override protected void run(JsonNode jsonNode) throws Exception {
         int backups = jsonNode.get("backups").asInt();
@@ -41,14 +49,40 @@ public class DataGenerationApplication extends IgniteAwareApplication {
         int from = jsonNode.get("from").asInt();
         int to = jsonNode.get("to").asInt();
 
+        int idxCnt = 0;
+
+        if (jsonNode.has("indexCount"))
+            idxCnt = jsonNode.get("indexCount").asInt();
+
         markInitialized();
 
         for (int i = 1; i <= cacheCnt; i++) {
-            IgniteCache<Integer, BinaryObject> cache = ignite.getOrCreateCache(
-                new CacheConfiguration<Integer, BinaryObject>("test-cache-" + i)
-                    .setBackups(backups));
+            CacheConfiguration<Integer, BinaryObject> ccfg = new CacheConfiguration<Integer, BinaryObject>("test-cache-" + i)
+                .setBackups(backups);
+
+            if (idxCnt > 0) {
+                QueryEntity qe = new QueryEntity();
+                List<QueryIndex> qi = new ArrayList<>(idxCnt);
+
+                qe.setKeyType(Integer.class.getName());
+                qe.setValueType(VAL_TYPE);
+
+                for (int j = 0; j < idxCnt; j++) {
+                    String field = "bytes" + j;
+
+                    qe.addQueryField(field, byte[].class.getName(), null);
+                    qi.add(new QueryIndex(field));
+                }
+
+                qe.setIndexes(qi);
 
-            generateCacheData(cache.getName(), entrySize, from, to);
+                ccfg.setQueryEntities(Collections.singleton(qe));
+
+            }
+
+            IgniteCache<Integer, BinaryObject> cache = ignite.getOrCreateCache(ccfg);
+
+            generateCacheData(cache.getName(), entrySize, from, to, idxCnt);
         }
 
         markFinished();
@@ -60,11 +94,11 @@ public class DataGenerationApplication extends IgniteAwareApplication {
      * @param from From key.
      * @param to To key.
      */
-    private void generateCacheData(String cacheName, int entrySize, int from, int to) {
+    private void generateCacheData(String cacheName, int entrySize, int from, int to, int idxCnt) {
         int flushEach = MAX_STREAMER_DATA_SIZE / entrySize + (MAX_STREAMER_DATA_SIZE % entrySize == 0 ? 0 : 1);
         int logEach = (to - from) / 10;
 
-        BinaryObjectBuilder builder = ignite.binary().builder("org.apache.ignite.ducktest.DataBinary");
+        BinaryObjectBuilder builder = ignite.binary().builder(VAL_TYPE);
 
         byte[] data = new byte[entrySize];
 
@@ -75,6 +109,14 @@ private void generateCacheData(String cacheName, int entrySize, int from, int to
                 builder.setField("key", i);
                 builder.setField("data", data);
 
+                for (int j = 0; j < idxCnt; j++) {
+                    byte[] indexedBytes = new byte[100];
+
+                    ThreadLocalRandom.current().nextBytes(indexedBytes);
+
+                    builder.setField("bytes" + j, indexedBytes);
+                }
+
                 stmr.addData(i, builder.build());
 
                 if ((i - from + 1) % logEach == 0 && log.isDebugEnabled())

modules/ducktests/tests/ignitetest/services/utils/path.py

@@ -215,6 +215,13 @@ def database_dir(self):
         """
         return os.path.join(self.work_dir, "db")
 
+    @property
+    def wal_dir(self):
+        """
+        :return: path to wal directory
+        """
+        return os.path.join(self.database_dir, "wal")
+
     @property
     def snapshots_dir(self):
         """
@@ -235,3 +242,20 @@ def script(self, script_name):
         :return: absolute path to the specified script
         """
         return os.path.join(self.home_dir, "bin", script_name)
+
+    def cache_dir(self, consistent_dir, cache_name):
+        """
+        :param consistent_dir: consistent ID directory.
+        :param cache_name: cache name.
+        :return: absolute path to the cache directory.
+        """
+        consistent_dir = consistent_dir.replace('.', '_')
+        return os.path.join(self.database_dir, consistent_dir, f'cache-{cache_name}')
+
+    def index_file(self, consistent_dir, cache_name):
+        """
+        :param consistent_dir: consistent ID directory.
+        :param cache_name: cache name.
+        :return: absolute path to the index file of cache.
+        """
+        return os.path.join(self.cache_dir(consistent_dir, cache_name), 'index.bin')

modules/ducktests/tests/ignitetest/tests/index_rebuild_test.py

@@ -0,0 +1,165 @@
+# 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.
+
+"""
+Module contains index.bin rebuild tests.
+"""
+import time
+
+from ducktape.mark import defaults
+
+from ignitetest.services.ignite import IgniteService
+from ignitetest.services.utils.control_utility import ControlUtility
+from ignitetest.services.utils.ignite_aware import IgniteAwareService
+from ignitetest.services.utils.ignite_configuration import DataStorageConfiguration, IgniteConfiguration
+from ignitetest.services.utils.ignite_configuration.data_storage import DataRegionConfiguration
+from ignitetest.services.utils.ignite_configuration.discovery import from_ignite_cluster
+from ignitetest.tests.rebalance.util import NUM_NODES
+from ignitetest.tests.util import DataGenerationParams, preload_data
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.ignite_test import IgniteTest
+from ignitetest.utils.version import DEV_BRANCH, LATEST, IgniteVersion
+
+
+def get_file_sizes(nodes: list, file: str) -> dict:
+    """
+    Return file size in bytes.
+
+    :param nodes: List of nodes.
+    :param file: File to get size for.
+    :return Dictionary with key as hostname value is files sizes on the node.
+    """
+    res = {}
+    for node in nodes:
+        out = IgniteAwareService.exec_command(node, f'du -sb {file}')
+
+        data = out.split("\t")
+
+        res[node.account.hostname] = int(data[0])
+
+    return res
+
+
+CACHE_NAME = "test-cache-1"
+
+
+class IndexRebuildTest(IgniteTest):
+    """
+    Tests index.bin rebuild.
+    """
+
+    @cluster(num_nodes=NUM_NODES)
+    @ignite_versions(str(DEV_BRANCH), str(LATEST))
+    @defaults(backups=[1], cache_count=[1], entry_count=[50000], entry_size=[50], preloaders=[1], index_count=[3])
+    def test_index_bin_rebuild(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
+                               index_count):
+        """
+        Tests index.bin rebuild on node start.
+        """
+
+        data_gen_params = DataGenerationParams(backups=backups, cache_count=cache_count, entry_count=entry_count,
+                                               entry_size=entry_size, preloaders=preloaders, index_count=index_count)
+
+        ignites = self.start_ignite(ignite_version, data_gen_params)
+
+        control_utility = ControlUtility(ignites)
+
+        control_utility.activate()
+
+        control_utility.disable_baseline_auto_adjust()
+
+        _, version, _ = control_utility.cluster_state()
+        control_utility.set_baseline(version)
+
+        preload_time = preload_data(
+            self.test_context,
+            ignites.config._replace(client_mode=True, discovery_spi=from_ignite_cluster(ignites)),
+            data_gen_params=data_gen_params)
+
+        control_utility.deactivate()
+
+        ignites.stop()
+
+        wal_before_rebuild = get_file_sizes(ignites.nodes, ignites.wal_dir)
+        idx_before_rebuild = get_file_sizes(ignites.nodes, ignites.index_file('*', CACHE_NAME))
+
+        for node in ignites.nodes:
+            IgniteAwareService.exec_command(node, f'rm {ignites.index_file(node.account.hostname, CACHE_NAME)}')
+
+        start_time = round(time.time() * 1000)
+
+        ignites.start(clean=False)
+
+        control_utility.activate()
+
+        timeout_sec = round(data_gen_params.entry_count / (len(ignites.nodes) * 250))
+
+        ignites.await_event(f"Started indexes rebuilding for cache \\[name={CACHE_NAME}, grpName=null\\]",
+                            from_the_beginning=True, timeout_sec=timeout_sec)
+        ignites.await_event("Indexes rebuilding completed for all caches.", from_the_beginning=True,
+                            timeout_sec=timeout_sec)
+
+        control_utility.deactivate()
+
+        ignites.stop()
+
+        rebuild_time = round(time.time() * 1000) - start_time
+
+        wal_after_rebuild = get_file_sizes(ignites.nodes, ignites.wal_dir)
+        idx_after_rebuild = get_file_sizes(ignites.nodes, ignites.index_file('*', CACHE_NAME))
+
+        wal_enlargement = {}
+        for node in wal_before_rebuild:
+            wal_enlargement[node] = wal_after_rebuild[node] - wal_before_rebuild[node]
+
+        return {
+            "preload_time": preload_time,
+            "wal_before_rebuild": wal_before_rebuild,
+            "wal_after_rebuild": wal_after_rebuild,
+            "idx_before_rebuild": idx_before_rebuild,
+            "idx_after_rebuild": idx_after_rebuild,
+            "wal_enlargement_bytes": wal_enlargement,
+            "rebuild_time_ms": rebuild_time
+        }
+
+    def start_ignite(self, ignite_version: str, data_gen_params: DataGenerationParams) -> IgniteService:
+        """
+        Start IgniteService:
+
+        :param ignite_version: Ignite version.
+        :param data_gen_params: Data generation parameters.
+        :return: IgniteService.
+        """
+        node_count = self.available_cluster_size - data_gen_params.preloaders
+
+        node_config = IgniteConfiguration(
+            cluster_state='INACTIVE',
+            auto_activation_enabled=False,
+            version=IgniteVersion(ignite_version),
+            data_storage=DataStorageConfiguration(
+                max_wal_archive_size=1000 * data_gen_params.data_region_max_size,
+                wal_segment_size=50 * 1024 * 1024,
+                default=DataRegionConfiguration(
+                    persistence_enabled=True,
+                    max_size=data_gen_params.data_region_max_size
+                )
+            ),
+            metric_exporters={"org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi"}
+        )
+
+        ignites = IgniteService(self.test_context, config=node_config, num_nodes=node_count)
+        ignites.start()
+
+        return ignites

modules/ducktests/tests/ignitetest/tests/util.py

@@ -34,6 +34,7 @@ class DataGenerationParams(NamedTuple):
     entry_count: int = 15_000
     entry_size: int = 50_000
     preloaders: int = 1
+    index_count: int = 0
 
     @property
     def data_region_max_size(self):
@@ -77,7 +78,8 @@ def start_app(_from, _to):
                 "cacheCount": data_gen_params.cache_count,
                 "entrySize": data_gen_params.entry_size,
                 "from": _from,
-                "to": _to
+                "to": _to,
+                "indexCount": data_gen_params.index_count
             },
             shutdown_timeout_sec=timeout)
         app.start_async()

modules/indexing/src/test/java/org/apache/ignite/internal/processors/database/WalDisabledDuringIndexRecreateTest.java

@@ -228,12 +228,14 @@ public void testRestartInCaseOfFailure() throws Exception {
     private void awaitRebuild() throws Exception {
         LogListener walDisabledLsnr = LogListener.matches(
             "WAL disabled for index partition " +
-                "[name=" + cacheGroupName() + ", id=" + cacheGroupId(cacheName(), cacheGroupName()) + ']'
+                "[name=" + (cacheGroupName() == null ? cacheName() : cacheGroupName()) +
+                ", id=" + cacheGroupId(cacheName(), cacheGroupName()) + ']'
         ).build();
 
         LogListener walEnabledLsnr = LogListener.matches(
             "WAL enabled for index partition " +
-                "[name=" + cacheGroupName() + ", id=" + cacheGroupId(cacheName(), cacheGroupName()) + ']'
+                "[name=" + (cacheGroupName() == null ? cacheName() : cacheGroupName()) +
+                ", id=" + cacheGroupId(cacheName(), cacheGroupName()) + ']'
         ).build();
 
         testLog.registerListener(walDisabledLsnr);