You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tl...@apache.org on 2021/06/17 06:42:15 UTC
[ignite] branch sql-calcite updated: IGNITE-13587 Additional
calcite tests with unstable topology (#9170)
This is an automated email from the ASF dual-hosted git repository.
tledkov pushed a commit to branch sql-calcite
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/sql-calcite by this push:
new 752624a IGNITE-13587 Additional calcite tests with unstable topology (#9170)
752624a is described below
commit 752624ac7913d541270450379c326a64b67e6fcc
Author: Evgeniy Stanilovskiy <st...@gmail.com>
AuthorDate: Thu Jun 17 09:41:41 2021 +0300
IGNITE-13587 Additional calcite tests with unstable topology (#9170)
---
.../query/calcite/UnstableTopologyTest.java | 206 +++++++++++++++++++++
.../ignite/testsuites/IntegrationTestSuite.java | 2 +
2 files changed, 208 insertions(+)
diff --git a/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/UnstableTopologyTest.java b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/UnstableTopologyTest.java
new file mode 100644
index 0000000..fa448fa
--- /dev/null
+++ b/modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/UnstableTopologyTest.java
@@ -0,0 +1,206 @@
+/*
+ * 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.processors.query.calcite;
+
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import com.google.common.collect.ImmutableList;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
+import org.apache.ignite.binary.BinaryObject;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.QueryEntity;
+import org.apache.ignite.cache.QueryIndex;
+import org.apache.ignite.cache.QueryIndexType;
+import org.apache.ignite.cluster.ClusterState;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
+import org.apache.ignite.internal.processors.query.GridQueryProcessor;
+import org.apache.ignite.internal.processors.query.GridQueryTypeDescriptor;
+import org.apache.ignite.internal.processors.query.QueryEngine;
+import org.apache.ignite.internal.processors.query.calcite.util.Commons;
+import org.apache.ignite.internal.processors.query.h2.IgniteH2Indexing;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+/** Non stable topology tests. */
+@RunWith(Parameterized.class)
+public class UnstableTopologyTest extends GridCommonAbstractTest {
+ /** */
+ private static final String POI_CACHE_NAME = "POI_CACHE";
+
+ /** */
+ private static final String POI_SCHEMA_NAME = "DOMAIN";
+
+ /** */
+ private static final String POI_TABLE_NAME = "POI";
+
+ /** */
+ private static final String POI_CLASS_NAME = "PointOfInterest";
+
+ /** */
+ private static final String ID_FIELD_NAME = "id";
+
+ /** */
+ private static final String NAME_FIELD_NAME = "name";
+
+ /** */
+ private static final String LATITUDE_FIELD_NAME = "latitude";
+
+ /** */
+ private static final String LONGITUDE_FIELD_NAME = "longitude";
+
+ /** */
+ private static final int NUM_ENTITIES = 2_000;
+
+ /** Test parameters. */
+ @Parameterized.Parameters(name = "awaitExchange={0}, idxSlowDown={1}")
+ public static List<Object[]> parameters() {
+ return ImmutableList.of(
+ new Object[]{true, true},
+ new Object[]{true, false},
+ new Object[]{false, false},
+ new Object[]{false, true}
+ );
+ }
+
+ /** */
+ @Parameterized.Parameter()
+ public boolean awaitExchange;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public boolean idxSlowDown;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setCacheConfiguration(new CacheConfiguration<>(POI_CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.ATOMIC)
+ .setSqlSchema(POI_SCHEMA_NAME)
+ .setQueryEntities(Collections.singletonList(queryEntity()))
+ .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setCacheMode(CacheMode.PARTITIONED)
+ .setIndexedTypes(Integer.class, Integer.class));
+
+ cfg.setClusterStateOnStart(ClusterState.ACTIVE);
+
+ return cfg;
+ }
+
+ /** */
+ private QueryEntity queryEntity() {
+ LinkedHashMap<String, String> fields = new LinkedHashMap<>();
+ fields.put(ID_FIELD_NAME, Integer.class.getName());
+ fields.put(NAME_FIELD_NAME, String.class.getName());
+ fields.put(LATITUDE_FIELD_NAME, Double.class.getName());
+ fields.put(LONGITUDE_FIELD_NAME, Double.class.getName());
+
+ return new QueryEntity()
+ .setKeyType(Integer.class.getName())
+ .setKeyFieldName(ID_FIELD_NAME)
+ .setValueType(POI_CLASS_NAME)
+ .setTableName(POI_TABLE_NAME)
+ .setFields(fields)
+ .setIndexes(Collections.singletonList(
+ new QueryIndex(NAME_FIELD_NAME, QueryIndexType.SORTED).setName(NAME_FIELD_NAME + "_idx")
+ ));
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids(true);
+
+ super.afterTest();
+ }
+
+ /**
+ * Checks sql execution correctness on changing (+1 node in parallel with request) topology.
+ */
+ @Test
+ public void testSelectCorrectnessOnUnstableTopology() throws Exception {
+ ignitionStart(0, idxSlowDown);
+ IgniteEx ig = ignitionStart(1, idxSlowDown);
+
+ loadData(ig, 0, NUM_ENTITIES);
+
+ ignitionStart(2, idxSlowDown);
+
+ if (awaitExchange)
+ awaitPartitionMapExchange(true, true, null);
+ else
+ awaitPartitionMapExchange();
+
+ QueryEngine engine = Commons.lookupComponent(grid(1).context(), QueryEngine.class);
+
+ G.allGrids().forEach(g -> assertEquals(NUM_ENTITIES, engine.query(null, POI_SCHEMA_NAME,
+ "SELECT * FROM " + POI_TABLE_NAME).get(0).getAll().size()));
+ }
+
+ /** */
+ private void loadData(Ignite node, int start, int end) {
+ try (IgniteDataStreamer<Object, Object> streamer = node.dataStreamer(POI_CACHE_NAME)) {
+ Random rnd = ThreadLocalRandom.current();
+
+ for (int i = start; i < end; i++) {
+ BinaryObject bo = node.binary().builder(POI_CLASS_NAME)
+ .setField(NAME_FIELD_NAME, "POI_" + i, String.class)
+ .setField(LATITUDE_FIELD_NAME, rnd.nextDouble(), Double.class)
+ .setField(LONGITUDE_FIELD_NAME, rnd.nextDouble(), Double.class)
+ .build();
+
+ streamer.addData(i, bo);
+ }
+ }
+ }
+
+ /** Start with custon QueryIndexing. */
+ private IgniteEx ignitionStart(int idx, boolean slow) throws Exception {
+ if (slow)
+ GridQueryProcessor.idxCls = BlockingIndexing.class;
+
+ return startGrid(getConfiguration(getTestIgniteInstanceName(idx)));
+ }
+
+ /**
+ * Simple blocking indexing processor.
+ */
+ private static class BlockingIndexing extends IgniteH2Indexing {
+ /** */
+ @Override public void remove(GridCacheContext cctx, GridQueryTypeDescriptor type, CacheDataRow row)
+ throws IgniteCheckedException {
+ U.sleep(50);
+
+ super.remove(cctx, type, row);
+ }
+ }
+}
diff --git a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
index a22299c..03fe094 100644
--- a/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
+++ b/modules/calcite/src/test/java/org/apache/ignite/testsuites/IntegrationTestSuite.java
@@ -25,6 +25,7 @@ import org.apache.ignite.internal.processors.query.calcite.DateTimeTest;
import org.apache.ignite.internal.processors.query.calcite.FunctionsTest;
import org.apache.ignite.internal.processors.query.calcite.LimitOffsetTest;
import org.apache.ignite.internal.processors.query.calcite.SqlFieldsQueryUsageTest;
+import org.apache.ignite.internal.processors.query.calcite.UnstableTopologyTest;
import org.apache.ignite.internal.processors.query.calcite.integration.AggregatesIntegrationTest;
import org.apache.ignite.internal.processors.query.calcite.integration.CalciteErrorHandlilngIntegrationTest;
import org.apache.ignite.internal.processors.query.calcite.integration.IndexSpoolIntegrationTest;
@@ -63,6 +64,7 @@ import org.junit.runners.Suite;
DataTypesTest.class,
IndexSpoolIntegrationTest.class,
SetOpIntegrationTest.class,
+ UnstableTopologyTest.class,
})
public class IntegrationTestSuite {
}