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/11/02 08:21:34 UTC
[ignite] branch master updated: IGNITE-15698 fix hang query when
table is dropped concurrently (unhandled exception at the
GridMapQueryExecutor#onQueryRequest) (#9506)
This is an automated email from the ASF dual-hosted git repository.
tledkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 5fe73f0 IGNITE-15698 fix hang query when table is dropped concurrently (unhandled exception at the GridMapQueryExecutor#onQueryRequest) (#9506)
5fe73f0 is described below
commit 5fe73f0826dcd71270bab7ce5ae7c6a86d550eb4
Author: Berkof <sa...@mail.ru>
AuthorDate: Tue Nov 2 15:21:00 2021 +0700
IGNITE-15698 fix hang query when table is dropped concurrently (unhandled exception at the GridMapQueryExecutor#onQueryRequest) (#9506)
---
.../spi/indexing/IndexingQueryFilterImpl.java | 7 ++
.../internal/processors/query/h2/H2Utils.java | 11 ++-
.../processors/query/h2/sql/GridSqlTable.java | 9 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 85 ++++++++++---------
...actQueryTableLockAndConnectionPoolSelfTest.java | 97 ++++++++++++++++++++++
5 files changed, 164 insertions(+), 45 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/indexing/IndexingQueryFilterImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/indexing/IndexingQueryFilterImpl.java
index 54fa554..04406f6 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/indexing/IndexingQueryFilterImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/indexing/IndexingQueryFilterImpl.java
@@ -21,6 +21,8 @@ import java.util.HashSet;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
+import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
+import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.util.typedef.F;
import org.jetbrains.annotations.Nullable;
@@ -94,6 +96,11 @@ public class IndexingQueryFilterImpl implements IndexingQueryFilter {
@Nullable @Override public IndexingQueryCacheFilter forCache(String cacheName) {
final GridCacheAdapter<Object, Object> cache = ctx.cache().internalCache(cacheName);
+ if (cache == null) {
+ throw new IgniteSQLException("Failed to find cache [cacheName=" + cacheName + ']',
+ IgniteQueryErrorCode.TABLE_NOT_FOUND);
+ }
+
// REPLICATED -> nothing to filter (explicit partitions are not supported).
if (cache.context().isReplicated() && !treatReplicatedAsPartitioned)
return null;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
index 356d46c..55d654e 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2Utils.java
@@ -41,6 +41,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
+
import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -909,7 +910,10 @@ public class H2Utils {
GridCacheContext cctx = sharedCtx.cacheContext(cacheId);
- assert cctx != null;
+ if (cctx == null) {
+ throw new IgniteSQLException("Failed to find cache [cacheId=" + cacheId + ']',
+ IgniteQueryErrorCode.TABLE_NOT_FOUND);
+ }
if (i == 0) {
mvccEnabled = cctx.mvccEnabled();
@@ -945,7 +949,10 @@ public class H2Utils {
GridCacheContext cctx = sharedCtx.cacheContext(cacheId);
- assert cctx != null;
+ if (cctx == null) {
+ throw new IgniteSQLException("Failed to find cache [cacheId=" + cacheId + ']',
+ IgniteQueryErrorCode.TABLE_NOT_FOUND);
+ }
if (!cctx.isPartitioned())
continue;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlTable.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlTable.java
index 02abc2d..315a944 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlTable.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/sql/GridSqlTable.java
@@ -19,6 +19,9 @@ package org.apache.ignite.internal.processors.query.h2.sql;
import java.util.Collections;
import java.util.List;
+
+import javax.cache.CacheException;
+
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.h2.command.Parser;
@@ -61,11 +64,11 @@ public class GridSqlTable extends GridSqlElement {
* @param tblName Table name.
* @param tbl H2 Table.
*/
- private GridSqlTable(@Nullable String schema, String tblName, @Nullable Table tbl) {
+ private GridSqlTable(String schema, String tblName, @Nullable Table tbl) {
super(Collections.<GridSqlAst>emptyList());
- assert schema != null : "schema";
- assert tblName != null : "tblName";
+ if (schema == null || tblName == null)
+ throw new CacheException("Table not found.");
this.schema = schema;
this.tblName = tblName;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index 6cdc51c..ccf967b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -217,33 +217,34 @@ public class GridMapQueryExecutor {
final boolean lazy = req.isFlagSet(GridH2QueryRequest.FLAG_LAZY);
boolean treatReplicatedAsPartitioned = req.isFlagSet(GridH2QueryRequest.FLAG_REPLICATED_AS_PARTITIONED);
- Boolean dataPageScanEnabled = req.isDataPageScanEnabled();
+ try {
+ Boolean dataPageScanEnabled = req.isDataPageScanEnabled();
- final List<Integer> cacheIds = req.caches();
+ final List<Integer> cacheIds = req.caches();
- final boolean singlePart = parts != null && parts.length == 1;
- final int parallelism = explain || replicated || F.isEmpty(cacheIds) ? 1 :
- CU.firstPartitioned(ctx.cache().context(), cacheIds).config().getQueryParallelism();
+ final boolean singlePart = parts != null && parts.length == 1;
+ final int parallelism = explain || replicated || F.isEmpty(cacheIds) ? 1 :
+ CU.firstPartitioned(ctx.cache().context(), cacheIds).config().getQueryParallelism();
- final int segments = explain || replicated || singlePart ? 1 : parallelism;
- final int singleSegment = singlePart ? calculateSegment(parallelism, parts[0]) : 0;
+ final int segments = explain || replicated || singlePart ? 1 : parallelism;
+ final int singleSegment = singlePart ? calculateSegment(parallelism, parts[0]) : 0;
- final Object[] params = req.parameters();
+ final Object[] params = req.parameters();
- final int timeout = req.timeout() > 0 || req.explicitTimeout()
- ? req.timeout()
- : (int)h2.distributedConfiguration().defaultQueryTimeout();
+ final int timeout = req.timeout() > 0 || req.explicitTimeout()
+ ? req.timeout()
+ : (int)h2.distributedConfiguration().defaultQueryTimeout();
- for (int i = 1; i < segments; i++) {
- assert !F.isEmpty(cacheIds);
+ for (int i = 1; i < segments; i++) {
+ assert !F.isEmpty(cacheIds);
- final int segment = i;
+ final int segment = i;
- Span span = MTC.span();
+ Span span = MTC.span();
- ctx.closure().callLocal(
- (GridPlainCallable<Void>)() -> {
- try (TraceSurroundings ignored = MTC.supportContinual(span)) {
+ ctx.closure().callLocal(
+ (GridPlainCallable<Void>)() -> {
+ try (TraceSurroundings ignored = MTC.supportContinual(span)) {
onQueryRequest0(node,
req.requestId(),
segment,
@@ -268,29 +269,33 @@ public class GridMapQueryExecutor {
return null;
}
},
- QUERY_POOL);
- }
+ QUERY_POOL);
+ }
- onQueryRequest0(node,
- req.requestId(),
- singleSegment,
- req.schemaName(),
- req.queries(),
- cacheIds,
- req.topologyVersion(),
- partsMap,
- parts,
- req.pageSize(),
- distributedJoins,
- enforceJoinOrder,
- replicated,
- timeout,
- params,
- lazy,
- req.mvccSnapshot(),
- dataPageScanEnabled,
- treatReplicatedAsPartitioned
- );
+ onQueryRequest0(node,
+ req.requestId(),
+ singleSegment,
+ req.schemaName(),
+ req.queries(),
+ cacheIds,
+ req.topologyVersion(),
+ partsMap,
+ parts,
+ req.pageSize(),
+ distributedJoins,
+ enforceJoinOrder,
+ replicated,
+ timeout,
+ params,
+ lazy,
+ req.mvccSnapshot(),
+ dataPageScanEnabled,
+ treatReplicatedAsPartitioned
+ );
+ }
+ catch (Throwable e) {
+ sendError(node, req.requestId(), e);
+ }
}
/**
diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/AbstractQueryTableLockAndConnectionPoolSelfTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/AbstractQueryTableLockAndConnectionPoolSelfTest.java
index 7b205ab..4090f1f 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/AbstractQueryTableLockAndConnectionPoolSelfTest.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/AbstractQueryTableLockAndConnectionPoolSelfTest.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicBoolean;
+import javax.cache.CacheException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
@@ -34,6 +35,7 @@ import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.processors.cache.index.AbstractIndexingCommonTest;
import org.apache.ignite.internal.processors.cache.query.SqlFieldsQueryEx;
import org.apache.ignite.internal.processors.query.h2.ConnectionManager;
@@ -198,6 +200,21 @@ public abstract class AbstractQueryTableLockAndConnectionPoolSelfTest extends Ab
}
/**
+ * Test drop table with high load queries on the same table.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testMultipleNodesWithTablesLockQueryAndTableDrop() throws Exception {
+ Ignite srv0 = startGrid(0);
+ Ignite srv1 = startGrid(1);
+ startGrid(2);
+
+ checkTablesLockQueryAndDropTable(srv0);
+ checkTablesLockQueryAndDropTable(srv1);
+ }
+
+ /**
* Test DDL operation on table with high load queries.
*
* @throws Exception If failed.
@@ -401,6 +418,86 @@ public abstract class AbstractQueryTableLockAndConnectionPoolSelfTest extends Ab
}
/**
+ * @param node Ignite node to execute query.
+ * @throws Exception If failed.
+ */
+ private void checkTablesLockQueryAndDropTable(final Ignite node) throws Exception {
+ execute(node, new SqlFieldsQuery("CREATE TABLE IF NOT EXISTS TEST (ID INT PRIMARY KEY, VAL INT)")).getAll();
+
+ final AtomicBoolean end = new AtomicBoolean(false);
+
+ final int qryThreads = 10;
+
+ // Do many concurrent queries.
+ IgniteInternalFuture<Long> fut = GridTestUtils.runMultiThreadedAsync(new Runnable() {
+ @Override public void run() {
+ while (!end.get()) {
+ try {
+ FieldsQueryCursor<List<?>> cursor = execute(node, new SqlFieldsQuery(
+ "SELECT * FROM TEST")
+ .setLazy(lazy()));
+
+ cursor.getAll();
+ }
+ catch (Exception e) {
+ String msg = e.getMessage();
+
+ if (msg != null && (msg.contains("Failed to find cache")
+ || msg.contains("Failed to perform cache operation (cache is stopped)")
+ || msg.contains("Failed to parse query. Table \"TEST\" not found")
+ || msg.contains("Cache not found on local node (was concurrently destroyed?)")
+ || msg.contains("Getting affinity for too old topology version that is already out of history")
+ || msg.contains("Failed to find partitioned cache")
+ || msg.contains("Table \"TEST\" not found")
+ || msg.contains("Table not found")
+ || msg.contains("Table PUBLIC.TEST already destroyed"))
+ ) {
+ // Swallow exception when table is dropped.
+ }
+ else if (X.cause(e, IgniteInterruptedCheckedException.class) != null) {
+ // Swallow exception when table is dropped.
+ }
+ //TODO: remove after https://issues.apache.org/jira/browse/IGNITE-15796
+ else if (X.cause(e, NullPointerException.class) != null) {
+ // Swallow exception when table is dropped.
+ }
+ else if (X.cause(e, CacheException.class) != null) {
+ // Swallow exception when table is dropped.
+ }
+ else if (X.cause(e, QueryRetryException.class) == null) {
+ log.error("Unexpected exception", e);
+
+ fail("Unexpected exception. " + e);
+ }
+ else if (!lazy()) {
+ log.error("Unexpected exception", e);
+
+ fail("Unexpected QueryRetryException.");
+ }
+ }
+ }
+ }
+ }, qryThreads, "usr-qry");
+
+ long tEnd = U.currentTimeMillis() + TEST_DUR;
+
+ while (U.currentTimeMillis() < tEnd) {
+ execute(node, new SqlFieldsQuery("DROP TABLE TEST")).getAll();
+
+ // Small delay after drop
+ U.sleep(10);
+
+ execute(node, new SqlFieldsQuery("CREATE TABLE TEST (ID INT PRIMARY KEY, VAL INT)")).getAll();
+ }
+
+ // Test is OK in case DDL operations is passed on hi load queries pressure.
+ end.set(true);
+ fut.get();
+
+ checkConnectionLeaks(Ignition.allGrids().size());
+ }
+
+ /**
* Test release reserved partition after query complete.
* In case partitions not released the `awaitPartitionMapExchange` fails by timeout.
*