You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pp...@apache.org on 2024/06/26 16:36:29 UTC
(ignite-3) branch main updated: IGNITE-22073 Sql. Add processing of the provided query timeout (#3953)
This is an automated email from the ASF dual-hosted git repository.
ppa 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 9b42e3e815 IGNITE-22073 Sql. Add processing of the provided query timeout (#3953)
9b42e3e815 is described below
commit 9b42e3e815b6c29daef9f8703bb5886f8f044973
Author: Max Zhuravkov <sh...@gmail.com>
AuthorDate: Wed Jun 26 19:36:24 2024 +0300
IGNITE-22073 Sql. Add processing of the provided query timeout (#3953)
---
.../java/org/apache/ignite/lang/ErrorGroups.java | 21 +-
.../jdbc/proto/event/JdbcBatchExecuteRequest.java | 25 +-
.../proto/event/JdbcBatchPreparedStmntRequest.java | 26 +-
.../jdbc/proto/event/JdbcQueryExecuteRequest.java | 29 +-
.../client/handler/JdbcQueryEventHandlerImpl.java | 43 ++-
.../handler/JdbcQueryEventHandlerImplTest.java | 14 +-
.../apache/ignite/internal/util/Cancellable.java | 4 +-
.../apache/ignite/jdbc/ItJdbcBatchSelfTest.java | 103 ++++++++
.../ignite/jdbc/ItJdbcMultiStatementSelfTest.java | 49 ++++
.../ignite/jdbc/ItJdbcStatementSelfTest.java | 67 +++++
.../internal/jdbc/JdbcPreparedStatement.java | 2 +-
.../apache/ignite/internal/jdbc/JdbcStatement.java | 33 ++-
modules/platforms/cpp/ignite/common/error_codes.h | 19 +-
modules/platforms/cpp/ignite/odbc/common_types.cpp | 2 -
.../dotnet/Apache.Ignite.Tests/FakeServer.cs | 2 +-
.../platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs | 21 +-
.../runner/app/ItIgniteNodeRestartTest.java | 3 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
.../ignite/internal/sql/api/ItSqlApiBaseTest.java | 99 +++++++
.../sql/engine/ItDynamicParameterTest.java | 38 ++-
.../ignite/internal/sql/api/IgniteSqlImpl.java | 8 +-
.../ignite/internal/sql/engine/QueryCancel.java | 107 ++++++--
.../sql/engine/QueryCancelledException.java | 22 +-
.../internal/sql/engine/SqlOperationContext.java | 6 +-
.../internal/sql/engine/SqlQueryProcessor.java | 73 ++++-
.../internal/sql/engine/exec/ExecutionContext.java | 11 +-
.../sql/engine/exec/ExecutionServiceImpl.java | 124 +++++++--
.../sql/engine/prepare/KeyValueGetPlan.java | 11 +
.../sql/engine/prepare/KeyValueModifyPlan.java | 11 +
.../sql/engine/prepare/PlanningContext.java | 2 +-
.../sql/engine/prepare/PrepareServiceImpl.java | 70 +++--
.../internal/sql/engine/util/cache/Cache.java | 14 +
.../engine/util/cache/CaffeineCacheFactory.java | 10 +
.../sql/engine/exec/ExecutionServiceImplTest.java | 294 ++++++++++++++++++++-
.../engine/exec/NoOpExecutableTableRegistry.java | 27 +-
.../sql/engine/exec/RuntimeSortedIndexTest.java | 3 +-
.../sql/engine/exec/rel/AbstractExecutionTest.java | 3 +-
.../engine/exec/rel/MergeJoinExecutionTest.java | 2 +-
.../sql/engine/framework/TestBuilders.java | 31 ++-
.../internal/sql/engine/framework/TestIndex.java | 15 +-
.../sql/engine/planner/AbstractPlannerTest.java | 2 +
.../sql/engine/planner/PlannerTimeoutTest.java | 4 +-
.../sql/engine/prepare/PrepareServiceImplTest.java | 101 ++++++-
.../sql/engine/util/EmptyCacheFactory.java | 10 +
.../internal/sql/engine/util/QueryCheckerImpl.java | 4 +-
45 files changed, 1368 insertions(+), 200 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 1c549389c1..c058a7eb75 100755
--- a/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -230,13 +230,13 @@ public class ErrorGroups {
public static final ErrorGroup SQL_ERR_GROUP = registerGroup("SQL", (short) 4);
/** Query without a result set error. */
- public static final int QUERY_NO_RESULT_SET_ERR = SQL_ERR_GROUP.registerErrorCode((short) 2);
+ public static final int QUERY_NO_RESULT_SET_ERR = SQL_ERR_GROUP.registerErrorCode((short) 1);
/** Schema not found. */
- public static final int SCHEMA_NOT_FOUND_ERR = SQL_ERR_GROUP.registerErrorCode((short) 3);
+ public static final int SCHEMA_NOT_FOUND_ERR = SQL_ERR_GROUP.registerErrorCode((short) 2);
/** Statement parsing error. This error is returned when an SQL statement string is not valid according to syntax rules. */
- public static final int STMT_PARSE_ERR = SQL_ERR_GROUP.registerErrorCode((short) 5);
+ public static final int STMT_PARSE_ERR = SQL_ERR_GROUP.registerErrorCode((short) 3);
/**
* Statement validation error. Although statement is grammatically correct, the semantic is in question.
@@ -250,13 +250,13 @@ public class ErrorGroups {
*
* <p>See message for details.
*/
- public static final int STMT_VALIDATION_ERR = SQL_ERR_GROUP.registerErrorCode((short) 6);
+ public static final int STMT_VALIDATION_ERR = SQL_ERR_GROUP.registerErrorCode((short) 4);
/** Constraint violation error such as primary key violation. */
- public static final int CONSTRAINT_VIOLATION_ERR = SQL_ERR_GROUP.registerErrorCode((short) 7);
+ public static final int CONSTRAINT_VIOLATION_ERR = SQL_ERR_GROUP.registerErrorCode((short) 5);
/** Statement canceled error. Statement is canceled due to timeout, admin action, etc. */
- public static final int EXECUTION_CANCELLED_ERR = SQL_ERR_GROUP.registerErrorCode((short) 8);
+ public static final int EXECUTION_CANCELLED_ERR = SQL_ERR_GROUP.registerErrorCode((short) 6);
/**
* Runtime error. Errors caused by programming errors in SQL statement itself, such errors happen during statement execution:
@@ -266,10 +266,7 @@ public class ErrorGroups {
* <li>Function execution errors.</li>
* </ul>
*/
- public static final int RUNTIME_ERR = SQL_ERR_GROUP.registerErrorCode((short) 9);
-
- /** Planning timed out without finding any valid plan. */
- public static final int PLANNING_TIMEOUT_ERR = SQL_ERR_GROUP.registerErrorCode((short) 10);
+ public static final int RUNTIME_ERR = SQL_ERR_GROUP.registerErrorCode((short) 7);
/**
* SQL engine was unable to map query on current cluster topology.
@@ -279,10 +276,10 @@ public class ErrorGroups {
*
* <p>See error message for details.
*/
- public static final int MAPPING_ERR = SQL_ERR_GROUP.registerErrorCode((short) 11);
+ public static final int MAPPING_ERR = SQL_ERR_GROUP.registerErrorCode((short) 8);
/** Execution of transaction control statement inside an external transaction is forbidden. */
- public static final int TX_CONTROL_INSIDE_EXTERNAL_TX_ERR = SQL_ERR_GROUP.registerErrorCode((short) 12);
+ public static final int TX_CONTROL_INSIDE_EXTERNAL_TX_ERR = SQL_ERR_GROUP.registerErrorCode((short) 9);
}
/** Meta storage error group. */
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
index 089ebb7317..c76179010d 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
@@ -38,6 +38,9 @@ public class JdbcBatchExecuteRequest implements ClientMessage {
/** Flag indicating whether auto-commit mode is enabled. */
private boolean autoCommit;
+ /** Query timeout in milliseconds. */
+ private long queryTimeoutMillis;
+
/**
* Default constructor.
*/
@@ -50,13 +53,20 @@ public class JdbcBatchExecuteRequest implements ClientMessage {
* @param schemaName Schema name.
* @param queries Queries.
* @param autoCommit Flag indicating whether auto-commit mode is enabled.
+ * @param queryTimeoutMillis Query timeout in millseconds.
*/
- public JdbcBatchExecuteRequest(String schemaName, List<String> queries, boolean autoCommit) {
+ public JdbcBatchExecuteRequest(
+ String schemaName,
+ List<String> queries,
+ boolean autoCommit,
+ long queryTimeoutMillis
+ ) {
assert !CollectionUtils.nullOrEmpty(queries);
this.schemaName = schemaName;
this.queries = queries;
this.autoCommit = autoCommit;
+ this.queryTimeoutMillis = queryTimeoutMillis;
}
/**
@@ -86,6 +96,15 @@ public class JdbcBatchExecuteRequest implements ClientMessage {
return autoCommit;
}
+ /**
+ * Returns the timeout in milliseconds.
+ *
+ * @return Timeout in milliseconds.
+ */
+ public long queryTimeoutMillis() {
+ return queryTimeoutMillis;
+ }
+
/** {@inheritDoc} */
@Override
public void writeBinary(ClientMessagePacker packer) {
@@ -97,6 +116,8 @@ public class JdbcBatchExecuteRequest implements ClientMessage {
for (String q : queries) {
packer.packString(q);
}
+
+ packer.packLong(queryTimeoutMillis);
}
/** {@inheritDoc} */
@@ -112,6 +133,8 @@ public class JdbcBatchExecuteRequest implements ClientMessage {
for (int i = 0; i < n; ++i) {
queries.add(unpacker.unpackString());
}
+
+ queryTimeoutMillis = unpacker.unpackLong();
}
/** {@inheritDoc} */
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
index a94093c4fe..46f5509ff8 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
@@ -42,6 +42,9 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
/** Flag indicating whether auto-commit mode is enabled. */
private boolean autoCommit;
+ /** Query timeout in milliseconds. */
+ private long queryTimeoutMillis;
+
/**
* Default constructor.
*/
@@ -55,8 +58,15 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
* @param query Sql query string.
* @param args Sql query arguments.
* @param autoCommit Flag indicating whether auto-commit mode is enabled.
+ * @param queryTimeoutMillis Query timeout in millseconds.
*/
- public JdbcBatchPreparedStmntRequest(String schemaName, String query, List<Object[]> args, boolean autoCommit) {
+ public JdbcBatchPreparedStmntRequest(
+ String schemaName,
+ String query,
+ List<Object[]> args,
+ boolean autoCommit,
+ long queryTimeoutMillis
+ ) {
assert !StringUtil.isNullOrEmpty(query);
assert !CollectionUtils.nullOrEmpty(args);
@@ -64,6 +74,7 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
this.args = args;
this.schemaName = schemaName;
this.autoCommit = autoCommit;
+ this.queryTimeoutMillis = queryTimeoutMillis;
}
/**
@@ -102,6 +113,15 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
return autoCommit;
}
+ /**
+ * Returns the timeout in milliseconds.
+ *
+ * @return Timeout in milliseconds.
+ */
+ public long queryTimeoutMillis() {
+ return queryTimeoutMillis;
+ }
+
/** {@inheritDoc} */
@Override
public void writeBinary(ClientMessagePacker packer) {
@@ -114,6 +134,8 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
for (Object[] arg : args) {
packer.packObjectArrayAsBinaryTuple(arg);
}
+
+ packer.packLong(queryTimeoutMillis);
}
/** {@inheritDoc} */
@@ -131,6 +153,8 @@ public class JdbcBatchPreparedStmntRequest implements ClientMessage {
for (int i = 0; i < n; ++i) {
args.add(unpacker.unpackObjectArrayFromBinaryTuple());
}
+
+ queryTimeoutMillis = unpacker.unpackLong();
}
/** {@inheritDoc} */
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
index de926c0a6c..4a521b0e7c 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
@@ -52,6 +52,9 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
/** Multiple statement flag. */
private boolean multiStatement;
+ /** Query timeout in milliseconds. */
+ private long queryTimeoutMillis;
+
/**
* Default constructor. For deserialization purposes.
*/
@@ -69,9 +72,19 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
* @param args Arguments list.
* @param autoCommit Flag indicating whether auto-commit mode is enabled.
* @param multiStatement Multiple statement flag.
+ * @param queryTimeoutMillis Query timeout in millseconds.
*/
- public JdbcQueryExecuteRequest(JdbcStatementType stmtType, String schemaName,
- int pageSize, int maxRows, String sqlQry, Object[] args, boolean autoCommit, boolean multiStatement) {
+ public JdbcQueryExecuteRequest(
+ JdbcStatementType stmtType,
+ String schemaName,
+ int pageSize,
+ int maxRows,
+ String sqlQry,
+ Object[] args,
+ boolean autoCommit,
+ boolean multiStatement,
+ long queryTimeoutMillis
+ ) {
Objects.requireNonNull(stmtType);
this.autoCommit = autoCommit;
@@ -82,6 +95,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
this.sqlQry = sqlQry;
this.args = args;
this.multiStatement = multiStatement;
+ this.queryTimeoutMillis = queryTimeoutMillis;
}
/**
@@ -154,6 +168,15 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
return autoCommit;
}
+ /**
+ * Returns the timeout in milliseconds.
+ *
+ * @return Timeout in milliseconds.
+ */
+ public long queryTimeoutMillis() {
+ return queryTimeoutMillis;
+ }
+
/** {@inheritDoc} */
@Override
public void writeBinary(ClientMessagePacker packer) {
@@ -166,6 +189,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
packer.packBoolean(multiStatement);
packer.packObjectArrayAsBinaryTuple(args);
+ packer.packLong(queryTimeoutMillis);
}
/** {@inheritDoc} */
@@ -180,6 +204,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
multiStatement = unpacker.unpackBoolean();
args = unpacker.unpackObjectArrayFromBinaryTuple();
+ queryTimeoutMillis = unpacker.unpackLong();
}
/** {@inheritDoc} */
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
index 3c4a3ab065..a73bad6915 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
@@ -142,7 +142,13 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
}
InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction();
- SqlProperties properties = createProperties(req.getStmtType(), req.multiStatement(), connectionContext.timeZoneId());
+
+ JdbcStatementType reqStmtType = req.getStmtType();
+ boolean multiStatement = req.multiStatement();
+ ZoneId timeZoneId = connectionContext.timeZoneId();
+ long timeoutMillis = req.queryTimeoutMillis();
+
+ SqlProperties properties = createProperties(reqStmtType, multiStatement, timeZoneId, timeoutMillis);
CompletableFuture<AsyncSqlCursor<InternalSqlRow>> result = processor.queryAsync(
properties,
@@ -156,7 +162,12 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
.exceptionally(t -> createErrorResult("Exception while executing query [query=" + req.sqlQuery() + "]", t, null));
}
- private static SqlProperties createProperties(JdbcStatementType stmtType, boolean multiStatement, ZoneId timeZoneId) {
+ private static SqlProperties createProperties(
+ JdbcStatementType stmtType,
+ boolean multiStatement,
+ ZoneId timeZoneId,
+ long queryTimeoutMillis
+ ) {
Set<SqlQueryType> allowedTypes;
switch (stmtType) {
@@ -176,6 +187,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
return SqlPropertiesHelper.newBuilder()
.set(QueryProperty.ALLOWED_QUERY_TYPES, allowedTypes)
.set(QueryProperty.TIME_ZONE_ID, timeZoneId)
+ .set(QueryProperty.QUERY_TIMEOUT, queryTimeoutMillis)
.build();
}
@@ -193,14 +205,19 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
var queries = req.queries();
var counters = new IntArrayList(req.queries().size());
var tail = CompletableFuture.completedFuture(counters);
+ long queryTimeoutMillis = req.queryTimeoutMillis();
for (String query : queries) {
- tail = tail.thenCompose(list -> executeAndCollectUpdateCount(connectionContext, tx, query, OBJECT_EMPTY_ARRAY)
- .thenApply(cnt -> {
- list.add(cnt > Integer.MAX_VALUE ? Statement.SUCCESS_NO_INFO : cnt.intValue());
-
- return list;
- }));
+ tail = tail.thenCompose(list -> executeAndCollectUpdateCount(
+ connectionContext,
+ tx, query,
+ OBJECT_EMPTY_ARRAY,
+ queryTimeoutMillis
+ ).thenApply(cnt -> {
+ list.add(cnt > Integer.MAX_VALUE ? Statement.SUCCESS_NO_INFO : cnt.intValue());
+
+ return list;
+ }));
}
return tail.handle((ignored, t) -> {
@@ -226,9 +243,10 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
var argList = req.getArgs();
var counters = new IntArrayList(req.getArgs().size());
var tail = CompletableFuture.completedFuture(counters);
+ long timeoutMillis = req.queryTimeoutMillis();
for (Object[] args : argList) {
- tail = tail.thenCompose(list -> executeAndCollectUpdateCount(connectionContext, tx, req.getQuery(), args)
+ tail = tail.thenCompose(list -> executeAndCollectUpdateCount(connectionContext, tx, req.getQuery(), args, timeoutMillis)
.thenApply(cnt -> {
list.add(cnt > Integer.MAX_VALUE ? Statement.SUCCESS_NO_INFO : cnt.intValue());
@@ -249,13 +267,14 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
JdbcConnectionContext context,
@Nullable InternalTransaction tx,
String sql,
- Object[] arg
- ) {
+ Object[] arg,
+ long timeoutMillis) {
+
if (!context.valid()) {
return CompletableFuture.failedFuture(new IgniteInternalException(CONNECTION_ERR, "Connection is closed"));
}
- SqlProperties properties = createProperties(JdbcStatementType.UPDATE_STATEMENT_TYPE, false, context.timeZoneId());
+ SqlProperties properties = createProperties(JdbcStatementType.UPDATE_STATEMENT_TYPE, false, context.timeZoneId(), timeoutMillis);
CompletableFuture<AsyncSqlCursor<InternalSqlRow>> result = processor.queryAsync(
properties,
diff --git a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
index b2a463adca..d26a857dee 100644
--- a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
+++ b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
@@ -126,7 +126,7 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
});
CompletableFuture<JdbcBatchExecuteResult> fut = IgniteTestUtils.runAsync(() ->
- eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("x", List.of("QUERY"), false)).get()
+ eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("x", List.of("QUERY"), false, 0)).get()
);
assertThat(registryCloseLatch.await(timeout, TimeUnit.SECONDS), is(true));
@@ -150,7 +150,7 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
long connectionId = acquireConnectionId();
- await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1"), false)));
+ await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1"), false, 0)));
verify(igniteTransactions).begin();
verify(tx, times(0)).rollbackAsync();
@@ -176,19 +176,19 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
String schema = "schema";
JdbcStatementType type = JdbcStatementType.SELECT_STATEMENT_TYPE;
- await(eventHandler.queryAsync(connectionId, new JdbcQueryExecuteRequest(type, schema, 1024, 1, "SELECT 1", null, false, false)));
+ await(eventHandler.queryAsync(connectionId, new JdbcQueryExecuteRequest(type, schema, 1024, 1, "SELECT 1", null, false, false, 0)));
verify(igniteTransactions, times(1)).begin();
- await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1", "UPDATE 2"), false)));
+ await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1", "UPDATE 2"), false, 0)));
verify(igniteTransactions, times(1)).begin();
await(eventHandler.finishTxAsync(connectionId, false));
verify(tx).rollbackAsync();
- await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1", "UPDATE 2"), false)));
+ await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 1", "UPDATE 2"), false, 0)));
verify(igniteTransactions, times(2)).begin();
- await(eventHandler.queryAsync(connectionId, new JdbcQueryExecuteRequest(type, schema, 1024, 1, "SELECT 2", null, false, false)));
+ await(eventHandler.queryAsync(connectionId, new JdbcQueryExecuteRequest(type, schema, 1024, 1, "SELECT 2", null, false, false, 0)));
verify(igniteTransactions, times(2)).begin();
- await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 3", "UPDATE 4"), false)));
+ await(eventHandler.batchAsync(connectionId, new JdbcBatchExecuteRequest("schema", List.of("UPDATE 3", "UPDATE 4"), false, 0)));
verify(igniteTransactions, times(2)).begin();
await(eventHandler.finishTxAsync(connectionId, true));
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/Cancellable.java b/modules/core/src/main/java/org/apache/ignite/internal/util/Cancellable.java
index a15e63598b..5f6cd721e1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/Cancellable.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/Cancellable.java
@@ -23,6 +23,8 @@ package org.apache.ignite.internal.util;
public interface Cancellable {
/**
* Cancels the ongoing operation or process or do nothing if it has been already cancelled.
+ *
+ * @param timeout If process was cancelled due to timeout.
*/
- void cancel();
+ void cancel(boolean timeout);
}
diff --git a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
index e531053e40..138666a5a4 100644
--- a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
+++ b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcBatchSelfTest.java
@@ -38,6 +38,9 @@ import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.Arrays;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.internal.jdbc.JdbcPreparedStatement;
+import org.apache.ignite.internal.jdbc.JdbcStatement;
import org.apache.ignite.internal.jdbc.proto.IgniteQueryErrorCode;
import org.apache.ignite.internal.jdbc.proto.SqlStateCode;
import org.junit.jupiter.api.AfterAll;
@@ -675,6 +678,106 @@ public class ItJdbcBatchSelfTest extends AbstractJdbcSelfTest {
assertEquals(0L, personsCount(), "Test table should be empty after empty batch is performed.");
}
+ @Test
+ public void testPreparedBatchTimeout() throws SQLException {
+ pstmt.close();
+
+ // Use some complex query that is not transformed into KV plan easily.
+ String updateStmt = "UPDATE Person SET age = ? "
+ + "WHERE id IN "
+ + "(SELECT * FROM TABLE(SYSTEM_RANGE(1, 50)) "
+ + "UNION "
+ + "SELECT * FROM TABLE(SYSTEM_RANGE(50, 100)))";
+ pstmt = conn.prepareStatement(updateStmt);
+
+ JdbcPreparedStatement igniteStmt = pstmt.unwrap(JdbcPreparedStatement.class);
+
+ {
+ // Disable timeout
+ igniteStmt.timeout(0);
+
+ for (int i = 0; i < 3; i++) {
+ pstmt.setInt(1, 42);
+
+ igniteStmt.addBatch();
+ }
+
+ int[] updated = igniteStmt.executeBatch();
+ assertEquals(3, updated.length);
+ }
+
+ // Each statement in a batch is executed separately, and timeout is applied to each statement.
+ {
+ int timeoutMillis = ThreadLocalRandom.current().nextInt(1, 5);
+ igniteStmt.timeout(timeoutMillis);
+
+ for (int i = 0; i < 3; i++) {
+ pstmt.setInt(1, 42);
+
+ igniteStmt.addBatch();
+ }
+
+ assertThrowsSqlException(SQLException.class,
+ "Query timeout", igniteStmt::executeBatch);
+ }
+
+ {
+ // Disable timeout
+ igniteStmt.timeout(0);
+
+ for (int i = 0; i < 3; i++) {
+ pstmt.setInt(1, 42);
+
+ igniteStmt.addBatch();
+ }
+
+ int[] updated = igniteStmt.executeBatch();
+ assertEquals(3, updated.length);
+ }
+ }
+
+ @Test
+ public void testBatchTimeout() throws SQLException {
+ JdbcStatement igniteStmt = stmt.unwrap(JdbcStatement.class);
+
+ {
+ // Disable timeout
+ igniteStmt.timeout(0);
+
+ for (int persIdx = 0; persIdx < 3; persIdx++) {
+ String stmt = "insert into Person (id, firstName, lastName, age) values " + generateValues(persIdx, 1);
+ igniteStmt.addBatch(stmt);
+ }
+
+ int[] updated = igniteStmt.executeBatch();
+ assertEquals(3, updated.length);
+ }
+
+ // Each statement in a batch is executed separately, so a timeout is applied to each statement.
+ {
+ igniteStmt.timeout(1);
+
+ String stmt = "insert into Person (id, firstName, lastName, age) values " + generateValues(200, 100);
+ igniteStmt.addBatch(stmt);
+
+ assertThrowsSqlException(SQLException.class,
+ "Query timeout", igniteStmt::executeBatch);
+ }
+
+ {
+ // Disable timeout
+ igniteStmt.timeout(0);
+
+ for (int persIdx = 10; persIdx < 13; persIdx++) {
+ String stmt = "insert into Person (id, firstName, lastName, age) values " + generateValues(persIdx, 1);
+ igniteStmt.addBatch(stmt);
+ }
+
+ int[] updated = igniteStmt.executeBatch();
+ assertEquals(3, updated.length);
+ }
+ }
+
/**
* Generate values for insert query.
*
diff --git a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultiStatementSelfTest.java b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultiStatementSelfTest.java
index 09c1164809..c96afd5054 100644
--- a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultiStatementSelfTest.java
+++ b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultiStatementSelfTest.java
@@ -20,12 +20,15 @@ package org.apache.ignite.jdbc;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.jdbc.util.JdbcTestUtils.assertThrowsSqlException;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import java.sql.Connection;
import java.sql.DriverManager;
@@ -33,6 +36,7 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import org.apache.ignite.internal.jdbc.JdbcStatement;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
@@ -647,6 +651,51 @@ public class ItJdbcMultiStatementSelfTest extends AbstractJdbcSelfTest {
}
}
+ @Test
+ public void testTimeout() throws SQLException {
+ JdbcStatement igniteStmt = (JdbcStatement) stmt;
+ igniteStmt.timeout(500);
+
+ int attempts = 10;
+
+ for (int i = 0; i < attempts; i++) {
+ stmt.execute("SELECT 1; SELECT * FROM TABLE(SYSTEM_RANGE(1, 1000000000)); SELECT * FROM TABLE(SYSTEM_RANGE(1, 10));");
+
+ // The first statement should succeed, if it times out, retry.
+ try {
+ try (ResultSet rs = stmt.getResultSet()) {
+ while (rs.next()) {
+ assertNotNull(rs.getObject(1));
+ }
+ }
+ } catch (SQLException e) {
+ // Ignore timeout for the first statement, if takes too long.
+ // Skip both planning and execution timeouts.
+ assertThat("Unexpected error", e.getMessage(), containsString("timeout"));
+ continue;
+ }
+
+ assertTrue(stmt.getMoreResults(), "Expected more results");
+
+ // The second statement should always fail.
+ assertThrowsSqlException(SQLException.class,
+ "Query timeout", () -> {
+ try (ResultSet rs = stmt.getResultSet()) {
+ while (rs.next()) {
+ assertNotNull(rs.getObject(1));
+ }
+ }
+ });
+
+ // Script timed out. We should also get a timeout.
+ assertThrowsSqlException(SQLException.class, "Query timeout", () -> stmt.getMoreResults());
+
+ return;
+ }
+
+ fail("Failed to get expected timeout error");
+ }
+
/**
* Extract person's name from result set.
*/
diff --git a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcStatementSelfTest.java b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcStatementSelfTest.java
index cf4acac6ba..f53e8befb6 100644
--- a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcStatementSelfTest.java
+++ b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcStatementSelfTest.java
@@ -19,6 +19,8 @@ package org.apache.ignite.jdbc;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
import static org.apache.ignite.jdbc.util.JdbcTestUtils.assertThrowsSqlException;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -35,6 +37,8 @@ import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.Statement;
import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.internal.jdbc.JdbcStatement;
import org.apache.ignite.sql.ColumnType;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
@@ -837,4 +841,67 @@ public class ItJdbcStatementSelfTest extends ItJdbcAbstractStatementSelfTest {
assertTrue(waitForCondition(() -> openCursors() == 0, 5_000));
}
+
+ @Test
+ public void testTimeout() throws Exception {
+ JdbcStatement igniteStmt = stmt.unwrap(JdbcStatement.class);
+
+ // No timeout
+
+ {
+ igniteStmt.timeout(0);
+
+ try (ResultSet rs = igniteStmt.executeQuery("SELECT * FROM TABLE(SYSTEM_RANGE(1, 100))")) {
+ while (rs.next()) {
+ rs.getLong(1);
+ }
+ }
+ }
+
+ // Rise timeout
+
+ {
+ int timeoutMillis = ThreadLocalRandom.current().nextInt(10, 200);
+ igniteStmt.timeout(timeoutMillis);
+
+ // Catch both execution and planning timeouts.
+ assertThrowsSqlException(SQLException.class,
+ "timeout", () -> {
+ try (ResultSet rs = igniteStmt.executeQuery("SELECT * FROM TABLE(SYSTEM_RANGE(1, 100000000))")) {
+ while (rs.next()) {
+ rs.getLong(1);
+ }
+ }
+ });
+ }
+
+ {
+ // Disable timeout
+
+ igniteStmt.timeout(0);
+
+ try (ResultSet rs = igniteStmt.executeQuery("SELECT * FROM TABLE(SYSTEM_RANGE(1, 100))")) {
+ while (rs.next()) {
+ rs.getLong(1);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testSetTimeoutValues() throws SQLException {
+ JdbcStatement igniteStmt = stmt.unwrap(JdbcStatement.class);
+
+ igniteStmt.timeout(1_234_000);
+ assertEquals(1_234, igniteStmt.getQueryTimeout());
+
+ igniteStmt.setQueryTimeout(Integer.MAX_VALUE);
+ assertEquals(Integer.MAX_VALUE, igniteStmt.getQueryTimeout());
+
+ igniteStmt.timeout(Integer.MAX_VALUE * 1000L);
+ assertEquals(Integer.MAX_VALUE, igniteStmt.getQueryTimeout());
+
+ SQLException err = assertThrows(SQLException.class, () -> igniteStmt.timeout(-1));
+ assertThat(err.getMessage(), containsString("Invalid timeout value"));
+ }
}
diff --git a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcPreparedStatement.java b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcPreparedStatement.java
index a38d9f2563..4cabfbb5ab 100644
--- a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcPreparedStatement.java
+++ b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcPreparedStatement.java
@@ -143,7 +143,7 @@ public class JdbcPreparedStatement extends JdbcStatement implements PreparedStat
}
JdbcBatchPreparedStmntRequest req
- = new JdbcBatchPreparedStmntRequest(conn.getSchema(), sql, batchedArgs, conn.getAutoCommit());
+ = new JdbcBatchPreparedStmntRequest(conn.getSchema(), sql, batchedArgs, conn.getAutoCommit(), queryTimeoutMillis);
try {
JdbcBatchExecuteResult res = conn.handler().batchPrepStatementAsync(conn.connectionId(), req).get();
diff --git a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
index 0729268247..aa8c48fdfb 100644
--- a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
+++ b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
@@ -70,7 +70,7 @@ public class JdbcStatement implements Statement {
private volatile boolean closed;
/** Query timeout. */
- private int timeout;
+ protected long queryTimeoutMillis;
/** Rows limit. */
private int maxRows;
@@ -137,7 +137,7 @@ public class JdbcStatement implements Statement {
}
JdbcQueryExecuteRequest req = new JdbcQueryExecuteRequest(stmtType, schema, pageSize, maxRows, sql, args,
- conn.getAutoCommit(), multiStatement);
+ conn.getAutoCommit(), multiStatement, queryTimeoutMillis);
JdbcQueryExecuteResponse res;
try {
@@ -286,21 +286,18 @@ public class JdbcStatement implements Statement {
public int getQueryTimeout() throws SQLException {
ensureNotClosed();
- return timeout / 1000;
+ long seconds = queryTimeoutMillis / 1000;
+ if (seconds >= Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ }
+
+ return (int) seconds;
}
/** {@inheritDoc} */
@Override
public void setQueryTimeout(int timeout) throws SQLException {
- ensureNotClosed();
-
- if (timeout < 0) {
- throw new SQLException("Invalid timeout value.");
- }
-
- // The timeout value of 0 will be converted to Integer.MAX_VALUE timeout to avoid further checks to 0.
- // This is because zero means there is no timeout limit.
- timeout(timeout * 1000 > timeout ? timeout * 1000 : Integer.MAX_VALUE);
+ timeout(timeout * 1000L);
}
/** {@inheritDoc} */
@@ -568,7 +565,7 @@ public class JdbcStatement implements Statement {
return INT_EMPTY_ARRAY;
}
- JdbcBatchExecuteRequest req = new JdbcBatchExecuteRequest(conn.getSchema(), batch, conn.getAutoCommit());
+ JdbcBatchExecuteRequest req = new JdbcBatchExecuteRequest(conn.getSchema(), batch, conn.getAutoCommit(), queryTimeoutMillis);
try {
JdbcBatchExecuteResult res = conn.handler().batchAsync(conn.connectionId(), req).get();
@@ -763,14 +760,16 @@ public class JdbcStatement implements Statement {
* <p>For test purposes.
*
* @param timeout Timeout.
- * @throws SQLException If timeout condition is not satisfied.
+ * @throws SQLException If timeout value is invalid.
*/
- public final void timeout(int timeout) throws SQLException {
+ public final void timeout(long timeout) throws SQLException {
+ ensureNotClosed();
+
if (timeout < 0) {
- throw new SQLException("Condition timeout >= 0 is not satisfied.");
+ throw new SQLException("Invalid timeout value.");
}
- this.timeout = timeout;
+ this.queryTimeoutMillis = timeout;
}
private static SQLException toSqlException(ExecutionException e) {
diff --git a/modules/platforms/cpp/ignite/common/error_codes.h b/modules/platforms/cpp/ignite/common/error_codes.h
index 2485639437..b46fd8ea09 100755
--- a/modules/platforms/cpp/ignite/common/error_codes.h
+++ b/modules/platforms/cpp/ignite/common/error_codes.h
@@ -94,16 +94,15 @@ enum class code : underlying_t {
HANDSHAKE_HEADER = 0x3000a,
// Sql group. Group code: 4
- QUERY_NO_RESULT_SET = 0x40002,
- SCHEMA_NOT_FOUND = 0x40003,
- STMT_PARSE = 0x40005,
- STMT_VALIDATION = 0x40006,
- CONSTRAINT_VIOLATION = 0x40007,
- EXECUTION_CANCELLED = 0x40008,
- RUNTIME = 0x40009,
- PLANNING_TIMEOUT = 0x4000a,
- MAPPING = 0x4000b,
- TX_CONTROL_INSIDE_EXTERNAL_TX = 0x4000c,
+ QUERY_NO_RESULT_SET = 0x40001,
+ SCHEMA_NOT_FOUND = 0x40002,
+ STMT_PARSE = 0x40003,
+ STMT_VALIDATION = 0x40004,
+ CONSTRAINT_VIOLATION = 0x40005,
+ EXECUTION_CANCELLED = 0x40006,
+ RUNTIME = 0x40007,
+ MAPPING = 0x40008,
+ TX_CONTROL_INSIDE_EXTERNAL_TX = 0x40009,
// MetaStorage group. Group code: 5
STARTING_STORAGE = 0x50001,
diff --git a/modules/platforms/cpp/ignite/odbc/common_types.cpp b/modules/platforms/cpp/ignite/odbc/common_types.cpp
index 9df2feaf15..82957f8ca0 100644
--- a/modules/platforms/cpp/ignite/odbc/common_types.cpp
+++ b/modules/platforms/cpp/ignite/odbc/common_types.cpp
@@ -158,8 +158,6 @@ sql_state error_code_to_sql_state(error::code code) {
// Sql group. Group code: 4
case error::code::SCHEMA_NOT_FOUND:
return sql_state::S3F000_INVALID_SCHEMA_NAME;
- case error::code::PLANNING_TIMEOUT:
- return sql_state::SHYT00_TIMEOUT_EXPIRED;
case error::code::EXECUTION_CANCELLED:
return sql_state::SHY008_OPERATION_CANCELED;
case error::code::TX_CONTROL_INSIDE_EXTERNAL_TX:
diff --git a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
index 9c0d0b80e7..8814df0774 100644
--- a/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
+++ b/modules/platforms/dotnet/Apache.Ignite.Tests/FakeServer.cs
@@ -383,7 +383,7 @@ namespace Apache.Ignite.Tests
using var errWriter = new PooledArrayBuffer();
var w = new MsgPackWriter(errWriter);
w.Write(Guid.Empty);
- w.Write(262150);
+ w.Write(262148);
w.Write("org.foo.bar.BazException");
w.Write(Err);
w.WriteNil(); // Stack trace.
diff --git a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
index 338a2a71ba..650d5bda20 100755
--- a/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
+++ b/modules/platforms/dotnet/Apache.Ignite/ErrorCodes.g.cs
@@ -177,34 +177,31 @@ namespace Apache.Ignite
public const String GroupName = "SQL";
/// <summary> QueryNoResultSet error. </summary>
- public const int QueryNoResultSet = (GroupCode << 16) | (2 & 0xFFFF);
+ public const int QueryNoResultSet = (GroupCode << 16) | (1 & 0xFFFF);
/// <summary> SchemaNotFound error. </summary>
- public const int SchemaNotFound = (GroupCode << 16) | (3 & 0xFFFF);
+ public const int SchemaNotFound = (GroupCode << 16) | (2 & 0xFFFF);
/// <summary> StmtParse error. </summary>
- public const int StmtParse = (GroupCode << 16) | (5 & 0xFFFF);
+ public const int StmtParse = (GroupCode << 16) | (3 & 0xFFFF);
/// <summary> StmtValidation error. </summary>
- public const int StmtValidation = (GroupCode << 16) | (6 & 0xFFFF);
+ public const int StmtValidation = (GroupCode << 16) | (4 & 0xFFFF);
/// <summary> ConstraintViolation error. </summary>
- public const int ConstraintViolation = (GroupCode << 16) | (7 & 0xFFFF);
+ public const int ConstraintViolation = (GroupCode << 16) | (5 & 0xFFFF);
/// <summary> ExecutionCancelled error. </summary>
- public const int ExecutionCancelled = (GroupCode << 16) | (8 & 0xFFFF);
+ public const int ExecutionCancelled = (GroupCode << 16) | (6 & 0xFFFF);
/// <summary> Runtime error. </summary>
- public const int Runtime = (GroupCode << 16) | (9 & 0xFFFF);
-
- /// <summary> PlanningTimeout error. </summary>
- public const int PlanningTimeout = (GroupCode << 16) | (10 & 0xFFFF);
+ public const int Runtime = (GroupCode << 16) | (7 & 0xFFFF);
/// <summary> Mapping error. </summary>
- public const int Mapping = (GroupCode << 16) | (11 & 0xFFFF);
+ public const int Mapping = (GroupCode << 16) | (8 & 0xFFFF);
/// <summary> TxControlInsideExternalTx error. </summary>
- public const int TxControlInsideExternalTx = (GroupCode << 16) | (12 & 0xFFFF);
+ public const int TxControlInsideExternalTx = (GroupCode << 16) | (9 & 0xFFFF);
}
/// <summary> MetaStorage errors. </summary>
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
index 9f592ee50a..31bc7074c7 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/ItIgniteNodeRestartTest.java
@@ -670,7 +670,8 @@ public class ItIgniteNodeRestartTest extends BaseIgniteRestartTest {
clusterConfigRegistry.getConfiguration(SqlDistributedConfiguration.KEY),
nodeCfgMgr.configurationRegistry().getConfiguration(SqlLocalConfiguration.KEY),
transactionInflights,
- txManager
+ txManager,
+ threadPoolsManager.commonScheduler()
);
sqlRef.set(new IgniteSqlImpl(qryEngine, new HybridTimestampTracker()));
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 41f529bcc5..f396fd3b57 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -878,7 +878,8 @@ public class IgniteImpl implements Ignite {
clusterConfigRegistry.getConfiguration(SqlDistributedConfiguration.KEY),
nodeConfigRegistry.getConfiguration(SqlLocalConfiguration.KEY),
transactionInflights,
- txManager
+ txManager,
+ threadPoolsManager.commonScheduler()
);
sql = new IgniteSqlImpl(qryEngine, observableTimestampTracker);
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlApiBaseTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlApiBaseTest.java
index 0ba0d825b9..39e156b1b4 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlApiBaseTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/api/ItSqlApiBaseTest.java
@@ -28,12 +28,14 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
@@ -66,6 +68,7 @@ import org.apache.ignite.tx.TransactionOptions;
import org.hamcrest.Matcher;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.AssertionFailureBuilder;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -894,6 +897,102 @@ public abstract class ItSqlApiBaseTest extends BaseSqlIntegrationTest {
assertEquals(nowMillis, tsMillis, deltaMillis);
}
+ @Test
+ public void testEarlyQueryTimeout() {
+ Statement stmt = igniteSql().statementBuilder()
+ .query("SELECT * FROM TABLE(SYSTEM_RANGE(1, 1000000000000000))")
+ .queryTimeout(1, TimeUnit.MILLISECONDS)
+ .build();
+
+ // Do not have enough time to do anything.
+ assertThrowsSqlException(Sql.EXECUTION_CANCELLED_ERR, "Query timeout", () -> {
+ executeForRead(igniteSql(), stmt);
+ });
+ }
+
+ @Test
+ public void testQueryTimeout() {
+ Statement stmt = igniteSql().statementBuilder()
+ .query("SELECT * FROM TABLE(SYSTEM_RANGE(1, 1000000000000000))")
+ .queryTimeout(100, TimeUnit.MILLISECONDS)
+ .build();
+
+ // Run ignoring any timeout util we get some result.
+ ResultSet<SqlRow> resultSet = runIgnoringExecutionErrors(stmt);
+ assertNotNull(resultSet);
+
+ // Read data until timeout exception occurs.
+ assertThrowsSqlException(Sql.EXECUTION_CANCELLED_ERR, "Query timeout", () -> {
+ while (resultSet.hasNext()) {
+ resultSet.next();
+ }
+ });
+ }
+
+ @Test
+ public void testQueryTimeoutIsPropagatedFromTheServer() throws Exception {
+ Statement stmt = igniteSql().statementBuilder()
+ .query("SELECT * FROM TABLE(SYSTEM_RANGE(1, 1000000000000000))")
+ .queryTimeout(100, TimeUnit.MILLISECONDS)
+ .build();
+
+ // Run ignoring any timeout to get some results.
+ ResultSet<SqlRow> resultSet = runIgnoringExecutionErrors(stmt);
+
+ assertTrue(resultSet.hasNext());
+ assertNotNull(resultSet.next());
+
+ // wait sometime until the time is right for a timeout to occur.
+ // then start retrieving the remaining data to trigger timeout exception.
+ TimeUnit.SECONDS.sleep(2);
+
+ assertThrowsSqlException(Sql.EXECUTION_CANCELLED_ERR, "Query timeout", () -> {
+ while (resultSet.hasNext()) {
+ resultSet.next();
+ }
+ });
+ }
+
+ private ResultSet<SqlRow> runIgnoringExecutionErrors(Statement stmt) {
+ SqlException lastError = null;
+
+ for (int i = 0; i < 100; i++) {
+ try {
+ return executeForRead(igniteSql(), stmt);
+ } catch (SqlException e) {
+ // Ignore all execution cancelled error. We assume that all these errors are transient (timeouts),
+ // and we will eventually get a result set.
+ if (e.code() == Sql.EXECUTION_CANCELLED_ERR) {
+ lastError = e;
+ continue;
+ }
+
+ fail(e.getMessage());
+ }
+ }
+
+ throw AssertionFailureBuilder.assertionFailure()
+ .message("Failed to execute the statement without timeouts.")
+ .cause(lastError)
+ .build();
+ }
+
+ @Test
+ public void testDdlTimeout() {
+ IgniteSql igniteSql = igniteSql();
+ int timeoutMillis = 1;
+
+ Statement stmt = igniteSql.statementBuilder()
+ .query("CREATE TABLE test (ID INT PRIMARY KEY, VAL0 INT)")
+ .queryTimeout(timeoutMillis, TimeUnit.MILLISECONDS)
+ .build();
+
+ // Trigger query timeout from the planner or the parser.
+ assertThrowsSqlException(Sql.EXECUTION_CANCELLED_ERR, "Query timeout", () -> {
+ executeForRead(igniteSql, stmt);
+ });
+ }
+
protected ResultSet<SqlRow> executeForRead(IgniteSql sql, String query, Object... args) {
return executeForRead(sql, null, query, args);
}
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItDynamicParameterTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItDynamicParameterTest.java
index d47c960beb..2e01078317 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItDynamicParameterTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItDynamicParameterTest.java
@@ -28,6 +28,7 @@ import java.sql.Date;
import java.time.LocalDate;
import java.util.Arrays;
import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.calcite.rel.type.RelDataType;
@@ -440,6 +441,40 @@ public class ItDynamicParameterTest extends BaseSqlIntegrationTest {
return ddl.toString();
}
+ @Test
+ public void testPrepareTimeout() {
+ StringBuilder stmt = new StringBuilder();
+
+ // Generate a complex query so it takes some to parse and validate.
+ // SELECT random-numbers
+ // UNION
+ // SELECT random-numbers
+ // ..
+ for (int i = 0; i < 5; i++) {
+ if (i > 0) {
+ stmt.append(System.lineSeparator());
+ stmt.append("UNION");
+ stmt.append(System.lineSeparator());
+ }
+
+ stmt.append("SELECT ");
+ for (int j = 0; j < 100; j++) {
+ stmt.append(ThreadLocalRandom.current().nextInt(0, 100));
+ stmt.append(", ");
+ }
+ stmt.setLength(stmt.length() - 2);
+ }
+
+ assertThrowsSqlException(Sql.EXECUTION_CANCELLED_ERR, "Query timeout", () -> {
+ SqlProperties properties = SqlPropertiesHelper.newBuilder()
+ .set(QueryProperty.QUERY_TIMEOUT, 1L)
+ .build();
+
+ QueryProcessor qryProc = queryProcessor();
+ await(qryProc.prepareSingleAsync(properties, null, stmt.toString())).parameterTypes();
+ });
+ }
+
@Override
protected int initialNodes() {
return 1;
@@ -454,7 +489,6 @@ public class ItDynamicParameterTest extends BaseSqlIntegrationTest {
private List<ParameterType> getParameterTypes(String query, Object... params) {
QueryProcessor qryProc = queryProcessor();
- SqlProperties properties = SqlPropertiesHelper.emptyProperties();
- return await(qryProc.prepareSingleAsync(properties, null, query, params)).parameterTypes();
+ return await(qryProc.prepareSingleAsync(SqlQueryProcessor.DEFAULT_PROPERTIES, null, query, params)).parameterTypes();
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/api/IgniteSqlImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/api/IgniteSqlImpl.java
index e0acc621a4..bb38cffd45 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/api/IgniteSqlImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/api/IgniteSqlImpl.java
@@ -52,6 +52,7 @@ import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.QueryProcessor;
import org.apache.ignite.internal.sql.engine.QueryProperty;
+import org.apache.ignite.internal.sql.engine.SqlQueryProcessor;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.property.SqlProperties;
import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
@@ -536,6 +537,8 @@ public class IgniteSqlImpl implements IgniteSql, IgniteComponent {
}
try {
+ SqlProperties properties = SqlQueryProcessor.DEFAULT_PROPERTIES;
+
return executeScriptCore(
queryProcessor,
observableTimestampTracker,
@@ -543,7 +546,7 @@ public class IgniteSqlImpl implements IgniteSql, IgniteComponent {
busyLock::leaveBusy,
query,
arguments,
- SqlPropertiesHelper.emptyProperties());
+ properties);
} finally {
busyLock.leaveBusy();
}
@@ -601,7 +604,8 @@ public class IgniteSqlImpl implements IgniteSql, IgniteComponent {
private static SqlProperties.Builder toPropertiesBuilder(Statement statement) {
return SqlPropertiesHelper.newBuilder()
.set(QueryProperty.TIME_ZONE_ID, statement.timeZoneId())
- .set(QueryProperty.DEFAULT_SCHEMA, statement.defaultSchema());
+ .set(QueryProperty.DEFAULT_SCHEMA, statement.defaultSchema())
+ .set(QueryProperty.QUERY_TIMEOUT, statement.queryTimeout(TimeUnit.MILLISECONDS));
}
private int registerCursor(AsyncSqlCursor<?> cursor) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancel.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancel.java
index 22fd77b25f..d6229ea49e 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancel.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancel.java
@@ -17,12 +17,17 @@
package org.apache.ignite.internal.sql.engine;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.util.Cancellable;
+import org.jetbrains.annotations.Nullable;
/**
* Holds query cancel state.
@@ -30,32 +35,104 @@ import org.apache.ignite.internal.util.Cancellable;
public class QueryCancel {
private final List<Cancellable> cancelActions = new ArrayList<>(3);
- private boolean canceled;
+ private Reason reason;
+
+ private volatile CompletableFuture<Void> timeoutFut;
/**
- * Adds a cancel action.
+ * Adds a cancel action. If operation has already been canceled, throws a {@link QueryCancelledException}.
+ *
+ * <p>NOTE: If the operation is cancelled, this method will immediately invoke the given action
+ * and then throw a {@link QueryCancelledException}.
*
- * @param clo Add cancel action.
+ * @param clo Cancel action.
*/
public synchronized void add(Cancellable clo) throws QueryCancelledException {
assert clo != null;
- if (canceled) {
- throw new QueryCancelledException();
+ if (reason != null) {
+ boolean timeout = reason == Reason.TIMEOUT;
+
+ // Immediately invoke a cancel action, if already cancelled.
+ // Otherwise the caller is required to catch QueryCancelledException and call an action manually.
+ try {
+ clo.cancel(timeout);
+ } catch (Exception ignore) {
+ // Do nothing
+ }
+
+ String message = timeout ? QueryCancelledException.TIMEOUT_MSG : QueryCancelledException.CANCEL_MSG;
+ throw new QueryCancelledException(message);
}
cancelActions.add(clo);
}
+ /**
+ * Removes the given callback.
+ *
+ * @param clo Callback.
+ */
+ public synchronized void remove(Cancellable clo) {
+ assert clo != null;
+
+ cancelActions.remove(clo);
+ }
+
+ /**
+ * Schedules a timeout after {@code timeoutMillis} milliseconds.
+ * Call be called only once.
+ *
+ * @param scheduler Scheduler to trigger an action.
+ * @param timeoutMillis Timeout in milliseconds.
+ * @return Future that will be completed when the timeout is reached.
+ */
+ public synchronized CompletableFuture<Void> setTimeout(ScheduledExecutorService scheduler, long timeoutMillis) {
+ assert reason == null : "Cannot set a timeout when cancelled";
+ assert timeoutFut == null : "Timeout has already been set";
+
+ CompletableFuture<Void> fut = new CompletableFuture<>();
+ fut.thenAccept((r) -> doCancel(Reason.TIMEOUT));
+
+ ScheduledFuture<?> f = scheduler.schedule(() -> {
+ fut.complete(null);
+ }, timeoutMillis, MILLISECONDS);
+
+ add((timeout) -> {
+ // Cancel the future if we didn't timeout,
+ // since in the case of a timeout it is already completed.
+ if (!timeout) {
+ f.cancel(false);
+ }
+ });
+
+ this.timeoutFut = fut;
+ return fut;
+ }
+
+ /**
+ * Returns the deadline of the operation.
+ *
+ * <p>Can be null if a query has no timeout.
+ */
+ public @Nullable CompletableFuture<Void> timeoutFuture() {
+ return timeoutFut;
+ }
+
/**
* Executes cancel closure.
*/
public synchronized void cancel() {
- if (canceled) {
+ doCancel(Reason.CANCEL);
+ }
+
+ private void doCancel(Reason reason) {
+ if (this.reason != null) {
return;
}
- canceled = true;
+ boolean timeout = reason == Reason.TIMEOUT;
+ this.reason = reason;
IgniteInternalException ex = null;
@@ -64,7 +141,7 @@ public class QueryCancel {
try {
Cancellable act = cancelActions.get(i);
- act.cancel();
+ act.cancel(timeout);
} catch (Exception e) {
if (ex == null) {
ex = new IgniteInternalException(INTERNAL_ERR, e);
@@ -79,16 +156,8 @@ public class QueryCancel {
}
}
- /**
- * Stops query execution if a user requested cancel.
- */
- public synchronized void checkCancelled() throws QueryCancelledException {
- if (canceled) {
- throw new QueryCancelledException();
- }
- }
-
- public synchronized boolean isCanceled() {
- return canceled;
+ enum Reason {
+ CANCEL,
+ TIMEOUT
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancelledException.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancelledException.java
index 427af0a840..7785373c33 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancelledException.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/QueryCancelledException.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.sql.engine;
import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
import org.apache.ignite.lang.IgniteException;
+import org.jetbrains.annotations.Nullable;
/**
* The exception is thrown if a query was cancelled or timed out while executing.
@@ -27,21 +28,32 @@ import org.apache.ignite.lang.IgniteException;
public class QueryCancelledException extends IgniteException {
private static final long serialVersionUID = 0L;
- public static final String ERR_MSG = "The query was cancelled while executing.";
+ public static final String CANCEL_MSG = "The query was cancelled while executing.";
+
+ public static final String TIMEOUT_MSG = "Query timeout";
/**
* Default constructor.
*/
public QueryCancelledException() {
- super(EXECUTION_CANCELLED_ERR, ERR_MSG);
+ super(EXECUTION_CANCELLED_ERR, CANCEL_MSG);
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param message Error message.
+ */
+ public QueryCancelledException(String message) {
+ super(EXECUTION_CANCELLED_ERR, message);
}
/**
* Constructor.
*
- * @param cause Optional nested exception (can be {@code null}).
+ * @param cause Cause.
*/
- public QueryCancelledException(Throwable cause) {
- super(EXECUTION_CANCELLED_ERR, ERR_MSG, cause);
+ public QueryCancelledException(@Nullable Throwable cause) {
+ super(EXECUTION_CANCELLED_ERR, CANCEL_MSG, cause);
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java
index 025aa66930..83f0d47692 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlOperationContext.java
@@ -52,7 +52,7 @@ public final class SqlOperationContext {
ZoneId timeZoneId,
Object[] parameters,
HybridTimestamp operationTime,
- QueryTransactionContext txContext,
+ @Nullable QueryTransactionContext txContext,
@Nullable QueryCancel cancel,
@Nullable String defaultSchemaName,
@Nullable PrefetchCallback prefetchCallback
@@ -85,7 +85,7 @@ public final class SqlOperationContext {
* Returns callback to notify about readiness of the first page of the results.
*
* <p>May be null on remote side, but never null on node initiator.
- */
+ */
public @Nullable PrefetchCallback prefetchCallback() {
return prefetchCallback;
}
@@ -141,7 +141,7 @@ public final class SqlOperationContext {
private ZoneId timeZoneId;
private Object[] parameters = ArrayUtils.OBJECT_EMPTY_ARRAY;
private HybridTimestamp operationTime;
- private QueryTransactionContext txContext;
+ private @Nullable QueryTransactionContext txContext;
private @Nullable QueryCancel cancel;
private @Nullable String defaultSchemaName;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
index 477496eb63..a7b3cae271 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueryProcessor.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.sql.engine;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.lang.SqlExceptionMapperUtil.mapToPublicSqlException;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -28,6 +29,8 @@ import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
+import java.time.Duration;
+import java.time.Instant;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
@@ -41,6 +44,8 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.LongSupplier;
import org.apache.ignite.internal.catalog.Catalog;
@@ -133,10 +138,12 @@ public class SqlQueryProcessor implements QueryProcessor {
/** Number of the schemas in cache. */
private static final int SCHEMA_CACHE_SIZE = 128;
- private static final SqlProperties DEFAULT_PROPERTIES = SqlPropertiesHelper.newBuilder()
+ /** Default properties. */
+ public static final SqlProperties DEFAULT_PROPERTIES = SqlPropertiesHelper.newBuilder()
.set(QueryProperty.DEFAULT_SCHEMA, SqlCommon.DEFAULT_SCHEMA_NAME)
.set(QueryProperty.ALLOWED_QUERY_TYPES, SqlQueryType.ALL)
.set(QueryProperty.TIME_ZONE_ID, DEFAULT_TIME_ZONE_ID)
+ .set(QueryProperty.QUERY_TIMEOUT, 0L)
.build();
private static final CacheFactory CACHE_FACTORY = CaffeineCacheFactory.INSTANCE;
@@ -208,6 +215,8 @@ public class SqlQueryProcessor implements QueryProcessor {
private final TransactionInflights transactionInflights;
+ private final ScheduledExecutorService commonScheduler;
+
/** Constructor. */
public SqlQueryProcessor(
ClusterService clusterSrvc,
@@ -227,7 +236,8 @@ public class SqlQueryProcessor implements QueryProcessor {
SqlDistributedConfiguration clusterCfg,
SqlLocalConfiguration nodeCfg,
TransactionInflights transactionInflights,
- TxManager txManager
+ TxManager txManager,
+ ScheduledExecutorService commonScheduler
) {
this.clusterSrvc = clusterSrvc;
this.logicalTopologyService = logicalTopologyService;
@@ -247,6 +257,7 @@ public class SqlQueryProcessor implements QueryProcessor {
this.nodeCfg = nodeCfg;
this.transactionInflights = transactionInflights;
this.txManager = txManager;
+ this.commonScheduler = commonScheduler;
sqlSchemaManager = new SqlSchemaManagerImpl(
catalogManager,
@@ -440,8 +451,12 @@ public class SqlQueryProcessor implements QueryProcessor {
) {
SqlProperties properties0 = SqlPropertiesHelper.chain(properties, DEFAULT_PROPERTIES);
String schemaName = properties0.get(QueryProperty.DEFAULT_SCHEMA);
+ Long queryTimeout = properties0.get(QueryProperty.QUERY_TIMEOUT);
QueryCancel queryCancel = new QueryCancel();
+ if (queryTimeout != 0) {
+ queryCancel.setTimeout(commonScheduler, queryTimeout);
+ }
ParsedResult parsedResult = queryToParsedResultCache.get(sql);
@@ -455,8 +470,20 @@ public class SqlQueryProcessor implements QueryProcessor {
HybridTimestamp timestamp = explicitTransaction != null ? explicitTransaction.startTimestamp() : clockService.now();
- return prepareParsedStatement(schemaName, result, timestamp, queryCancel, params)
+ CompletableFuture<QueryMetadata> f = prepareParsedStatement(schemaName, result, timestamp,
+ queryCancel, params)
.thenApply(plan -> new QueryMetadata(plan.metadata(), plan.parameterMetadata()));
+
+ CompletableFuture<Void> timeoutFut = queryCancel.timeoutFuture();
+
+ if (timeoutFut != null) {
+ timeoutFut.thenAccept((r) -> {
+ SqlException timeoutErr = new SqlException(EXECUTION_CANCELLED_ERR, QueryCancelledException.TIMEOUT_MSG);
+ f.completeExceptionally(timeoutErr);
+ });
+ }
+
+ return f;
});
}
@@ -468,9 +495,14 @@ public class SqlQueryProcessor implements QueryProcessor {
) {
String schemaName = properties.get(QueryProperty.DEFAULT_SCHEMA);
ZoneId timeZoneId = properties.get(QueryProperty.TIME_ZONE_ID);
+ Long queryTimeout = properties.get(QueryProperty.QUERY_TIMEOUT);
QueryCancel queryCancel = new QueryCancel();
+ if (queryTimeout != 0) {
+ queryCancel.setTimeout(commonScheduler, queryTimeout);
+ }
+
ParsedResult parsedResult = queryToParsedResultCache.get(sql);
CompletableFuture<ParsedResult> start = parsedResult != null
@@ -506,6 +538,14 @@ public class SqlQueryProcessor implements QueryProcessor {
) {
String schemaName = properties.get(QueryProperty.DEFAULT_SCHEMA);
ZoneId timeZoneId = properties.get(QueryProperty.TIME_ZONE_ID);
+ Long queryTimeout = properties.get(QueryProperty.QUERY_TIMEOUT);
+
+ Instant deadline;
+ if (queryTimeout != 0) {
+ deadline = Instant.now().plusMillis(queryTimeout);
+ } else {
+ deadline = null;
+ }
CompletableFuture<?> start = new CompletableFuture<>();
@@ -513,7 +553,7 @@ public class SqlQueryProcessor implements QueryProcessor {
.thenApply(ignored -> parserService.parseScript(sql))
.thenCompose(parsedResults -> {
MultiStatementHandler handler = new MultiStatementHandler(
- schemaName, txCtx, parsedResults, params, timeZoneId);
+ schemaName, txCtx, parsedResults, params, timeZoneId, deadline);
return handler.processNext();
});
@@ -743,6 +783,7 @@ public class SqlQueryProcessor implements QueryProcessor {
private final String schemaName;
private final Queue<ScriptStatement> statements;
private final ScriptTransactionContext scriptTxContext;
+ private final @Nullable Instant deadline;
/**
* Collection is used to track SELECT statements to postpone following DML operation.
@@ -760,12 +801,14 @@ public class SqlQueryProcessor implements QueryProcessor {
QueryTransactionContext txContext,
List<ParsedResult> parsedResults,
Object[] params,
- ZoneId timeZoneId
+ ZoneId timeZoneId,
+ @Nullable Instant deadline
) {
this.timeZoneId = timeZoneId;
this.schemaName = schemaName;
this.statements = prepareStatementsQueue(parsedResults, params);
this.scriptTxContext = new ScriptTransactionContext(txContext, transactionInflights);
+ this.deadline = deadline;
}
/**
@@ -834,14 +877,32 @@ public class SqlQueryProcessor implements QueryProcessor {
new IteratorToDataCursorAdapter<>(Collections.emptyIterator()),
nextCurFut
));
+
+ if (deadline != null) {
+ long statementTimeoutMillis = Duration.between(Instant.now(), deadline).toMillis();
+
+ ScheduledFuture<?> f = commonScheduler.schedule(() -> {
+ SqlException err = new SqlException(EXECUTION_CANCELLED_ERR, QueryCancelledException.TIMEOUT_MSG);
+ fut.completeExceptionally(err);
+ }, statementTimeoutMillis, MILLISECONDS);
+
+ fut.whenComplete((r, t) -> f.cancel(false));
+ }
} else {
scriptTxContext.registerCursorFuture(parsedResult.queryType(), cursorFuture);
HybridTimestamp operationTime = deriveOperationTime(scriptTxContext);
+ QueryCancel queryCancel = new QueryCancel();
+
+ if (deadline != null) {
+ long statementTimeoutMillis = Duration.between(Instant.now(), deadline).toMillis();
+ queryCancel.setTimeout(commonScheduler, statementTimeoutMillis);
+ }
+
SqlOperationContext operationContext = SqlOperationContext.builder()
.queryId(UUID.randomUUID())
- .cancel(new QueryCancel())
+ .cancel(queryCancel)
.prefetchCallback(new PrefetchCallback())
.parameters(params)
.timeZoneId(timeZoneId)
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
index 512ace5503..02cf1e7f42 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
@@ -94,6 +94,8 @@ public class ExecutionContext<RowT> implements DataContext {
private SharedState sharedState = new SharedState();
+ private final @Nullable CompletableFuture<Void> timeoutFut;
+
/**
* Constructor.
*
@@ -106,6 +108,7 @@ public class ExecutionContext<RowT> implements DataContext {
* @param params Parameters.
* @param txAttributes Transaction attributes.
* @param timeZoneId Session time-zone ID.
+ * @param timeoutFut Timeout future.
*/
@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
public ExecutionContext(
@@ -117,7 +120,8 @@ public class ExecutionContext<RowT> implements DataContext {
RowHandler<RowT> handler,
Map<String, Object> params,
TxAttributes txAttributes,
- ZoneId timeZoneId
+ ZoneId timeZoneId,
+ @Nullable CompletableFuture<Void> timeoutFut
) {
this.executor = executor;
this.qryId = qryId;
@@ -128,6 +132,7 @@ public class ExecutionContext<RowT> implements DataContext {
this.originatingNodeName = originatingNodeName;
this.txAttributes = txAttributes;
this.timeZoneId = timeZoneId;
+ this.timeoutFut = timeoutFut;
expressionFactory = new ExpressionFactoryImpl<>(
this,
@@ -382,6 +387,10 @@ public class ExecutionContext<RowT> implements DataContext {
return cancelFlag.get();
}
+ public @Nullable CompletableFuture<Void> timeoutFuture() {
+ return timeoutFut;
+ }
+
/** Creates {@link PartitionProvider} for the given source table. */
public PartitionProvider<RowT> getPartitionProvider(long sourceId, ColocationGroup group, IgniteTable table) {
PartitionPruningMetadata metadata = description.partitionPruningMetadata();
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index be4d7705f6..c314582b68 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -119,6 +119,7 @@ import org.apache.ignite.internal.util.AsyncCursor;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.network.ClusterNode;
import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.TestOnly;
/**
* Provide ability to execute SQL query plan and retrieve results of the execution.
@@ -263,7 +264,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
exchangeSrvc,
deps,
tableFunctionRegistry
- ),
+ ),
clockService,
shutdownTimeout
);
@@ -278,6 +279,16 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
messageService.register((n, m) -> onMessage(n, (ErrorMessage) m), SqlQueryMessageGroup.ERROR_MESSAGE);
}
+ @TestOnly
+ public ExecutableTableRegistry tableRegistry() {
+ return tableRegistry;
+ }
+
+ @TestOnly
+ public DdlCommandHandler ddlCommandHandler() {
+ return ddlCmdHnd;
+ }
+
private AsyncDataCursor<InternalSqlRow> executeQuery(
SqlOperationContext operationContext,
MultiStepPlan plan
@@ -292,7 +303,16 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
assert cancelHandler != null;
- cancelHandler.add(() -> queryManager.close(true));
+ // This call triggers a timeout exception, if operation has timed out.
+ cancelHandler.add(timeout -> {
+ QueryCompletionReason reason = timeout ? QueryCompletionReason.TIMEOUT : QueryCompletionReason.CANCEL;
+ queryManager.close(reason);
+ });
+
+ CompletableFuture<Void> timeoutFut = cancelHandler.timeoutFuture();
+ if (timeoutFut != null) {
+ timeoutFut.thenAcceptAsync((r) -> queryManager.close(QueryCompletionReason.TIMEOUT), taskExecutor);
+ }
QueryTransactionContext txContext = operationContext.txContext();
@@ -361,9 +381,9 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return executeQuery(operationContext, (MultiStepPlan) plan);
case EXPLAIN:
- return executeExplain((ExplainPlan) plan, operationContext.prefetchCallback());
+ return executeExplain(operationContext, (ExplainPlan) plan);
case DDL:
- return executeDdl((DdlPlan) plan, operationContext.prefetchCallback());
+ return executeDdl(operationContext, (DdlPlan) plan);
default:
throw new AssertionError("Unexpected query type: " + plan);
@@ -378,13 +398,16 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return nullCompletedFuture();
}
- return mgr.close(true);
+ return mgr.close(QueryCompletionReason.CANCEL);
}
private AsyncDataCursor<InternalSqlRow> executeExecutablePlan(
SqlOperationContext operationContext,
ExecutablePlan plan
) {
+ QueryCancel queryCancel = operationContext.cancel();
+ assert queryCancel != null;
+
ExecutionContext<RowT> ectx = new ExecutionContext<>(
taskExecutor,
operationContext.queryId(),
@@ -394,7 +417,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
handler,
Commons.parametersMap(operationContext.parameters()),
TxAttributes.dummy(),
- operationContext.timeZoneId()
+ operationContext.timeZoneId(),
+ queryCancel.timeoutFuture()
);
QueryTransactionContext txContext = operationContext.txContext();
@@ -422,17 +446,30 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
);
}
- private AsyncDataCursor<InternalSqlRow> executeDdl(DdlPlan plan, @Nullable QueryPrefetchCallback callback) {
+ private AsyncDataCursor<InternalSqlRow> executeDdl(SqlOperationContext operationContext,
+ DdlPlan plan
+ ) {
CompletableFuture<Iterator<InternalSqlRow>> ret = ddlCmdHnd.handle(plan.command())
.thenApply(applied -> (applied ? APPLIED_ANSWER : NOT_APPLIED_ANSWER).iterator())
.exceptionally(th -> {
throw convertDdlException(th);
});
+ PrefetchCallback callback = operationContext.prefetchCallback();
if (callback != null) {
ret.whenCompleteAsync((res, err) -> callback.onPrefetchComplete(err), taskExecutor);
}
+ QueryCancel queryCancel = operationContext.cancel();
+ assert queryCancel != null;
+
+ CompletableFuture<Void> timeoutFut = queryCancel.timeoutFuture();
+ if (timeoutFut != null) {
+ timeoutFut.whenCompleteAsync((r, t) -> {
+ ret.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG));
+ }, taskExecutor);
+ }
+
return new IteratorToDataCursorAdapter<>(ret, Runnable::run);
}
@@ -453,7 +490,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return (e instanceof RuntimeException) ? (RuntimeException) e : new IgniteInternalException(INTERNAL_ERR, e);
}
- private AsyncDataCursor<InternalSqlRow> executeExplain(ExplainPlan plan, @Nullable QueryPrefetchCallback callback) {
+ private AsyncDataCursor<InternalSqlRow> executeExplain(SqlOperationContext operationContext, ExplainPlan plan) {
+ QueryPrefetchCallback callback = operationContext.prefetchCallback();
String planString = plan.plan().explain();
InternalSqlRow res = new InternalSqlRowSingleString(planString);
@@ -523,7 +561,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
DistributedQueryManager dqm = queryManagerMap.get(msg.queryId());
if (dqm != null) {
- dqm.close(true);
+ dqm.close(QueryCompletionReason.CANCEL);
}
}
@@ -532,7 +570,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
public void stop() throws Exception {
CompletableFuture<Void> f = CompletableFuture.allOf(queryManagerMap.values().stream()
.filter(mgr -> mgr.rootFragmentId != null)
- .map(mgr -> mgr.close(true))
+ .map(mgr -> mgr.close(QueryCompletionReason.CANCEL))
.toArray(CompletableFuture[]::new)
);
@@ -736,6 +774,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private final @Nullable CompletableFuture<AsyncRootNode<RowT, InternalSqlRow>> root;
+ private final @Nullable CompletableFuture<Void> timeoutFut;
+
private volatile Long rootFragmentId = null;
private DistributedQueryManager(
@@ -748,10 +788,13 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
this.coordinatorNodeName = coordinatorNodeName;
if (coordinator) {
+ QueryCancel queryCancel = ctx.cancel();
+ assert queryCancel != null;
+
var root = new CompletableFuture<AsyncRootNode<RowT, InternalSqlRow>>();
root.exceptionally(t -> {
- this.close(true);
+ this.close(QueryCompletionReason.ERROR);
QueryPrefetchCallback callback = ctx.prefetchCallback();
@@ -763,8 +806,10 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
});
this.root = root;
+ this.timeoutFut = queryCancel.timeoutFuture();
} else {
this.root = null;
+ this.timeoutFut = null;
}
}
@@ -801,23 +846,31 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
if (rootFragmentId0 != null && fragmentId == rootFragmentId0) {
root.completeExceptionally(ex);
+ } else if (root == null) {
+ // Non-root fragment received an error when attempted to submit a fragment.
+ close(QueryCompletionReason.ERROR);
} else {
root.thenAccept(root -> {
root.onError(ex);
- close(true);
+ close(QueryCompletionReason.ERROR);
});
}
}
- remoteFragmentInitCompletion.get(new RemoteFragmentKey(nodeName, fragmentId)).complete(null);
+ // Complete if fragment is registered.
+ // TODO https://issues.apache.org/jira/browse/IGNITE-22585
+ CompletableFuture<Void> f = remoteFragmentInitCompletion.get(new RemoteFragmentKey(nodeName, fragmentId));
+ if (f != null) {
+ f.complete(null);
+ }
}
private void onError(RemoteFragmentExecutionException ex) {
root.thenAccept(root -> {
root.onError(ex);
- close(true);
+ close(QueryCompletionReason.ERROR);
});
}
@@ -877,7 +930,8 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
handler,
Commons.parametersMap(ctx.parameters()),
txAttributes,
- ctx.timeZoneId()
+ ctx.timeZoneId(),
+ timeoutFut
);
}
@@ -923,7 +977,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
} catch (Exception e) {
LOG.info("Unable to send error message", e);
- close(true);
+ close(QueryCompletionReason.ERROR);
}
}
@@ -1068,7 +1122,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
fut.thenAccept(batch -> {
if (!batch.hasMore()) {
- DistributedQueryManager.this.close(false);
+ DistributedQueryManager.this.close();
}
});
@@ -1085,7 +1139,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return DistributedQueryManager.this.cancelFut;
}
- return DistributedQueryManager.this.close(false);
+ return DistributedQueryManager.this.close();
}).thenCompose(Function.identity());
}
};
@@ -1152,7 +1206,11 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
}.visit(mappedFragment.fragment().root());
}
- private CompletableFuture<Void> close(boolean cancel) {
+ private CompletableFuture<Void> close() {
+ return close(QueryCompletionReason.CLOSE);
+ }
+
+ private CompletableFuture<Void> close(QueryCompletionReason reason) {
if (!cancelled.compareAndSet(false, true)) {
return cancelFut;
}
@@ -1162,7 +1220,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
CompletableFuture<Void> stage;
if (coordinator) {
- stage = start.thenCompose(ignored -> closeRootNode(cancel))
+ stage = start.thenCompose(ignored -> closeRootNode(reason))
.thenCompose(ignored -> awaitFragmentInitialisationAndClose());
} else {
stage = start.thenCompose(ignored -> messageService.send(coordinatorNodeName, FACTORY.queryCloseMessage()
@@ -1182,6 +1240,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
QueryCancel cancelHandler = ctx.cancel();
+ // Query cancel runs only at the coordinator node.
if (cancelHandler != null) {
try {
cancelHandler.cancel();
@@ -1255,21 +1314,28 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
/**
* Synchronously closes the tree's execution iterator.
*
- * @param cancel Forces execution to terminate with {@link QueryCancelledException}.
+ * @param closeReason Reason to use in {@link QueryCancelledException}.
* @return Completable future that should run asynchronously.
*/
- private CompletableFuture<Void> closeRootNode(boolean cancel) {
+ private CompletableFuture<Void> closeRootNode(QueryCompletionReason closeReason) {
assert root != null;
+ String message;
+ if (closeReason == QueryCompletionReason.TIMEOUT) {
+ message = QueryCancelledException.TIMEOUT_MSG;
+ } else {
+ message = QueryCancelledException.CANCEL_MSG;
+ }
+
if (!root.isDone()) {
- root.completeExceptionally(new QueryCancelledException());
+ root.completeExceptionally(new QueryCancelledException(message));
}
if (!root.isCompletedExceptionally()) {
AsyncRootNode<RowT, InternalSqlRow> node = root.getNow(null);
- if (cancel) {
- node.onError(new QueryCancelledException());
+ if (closeReason != QueryCompletionReason.CLOSE) {
+ node.onError(new QueryCancelledException(message));
}
return node.closeAsync();
@@ -1321,4 +1387,12 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
return Objects.hash(catalogVersion, fragmentString);
}
}
+
+ /** Represents reasons why a query was completed. */
+ private enum QueryCompletionReason {
+ CLOSE,
+ CANCEL,
+ TIMEOUT,
+ ERROR,
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
index 1489dd6b79..8beffb1e22 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueGetPlan.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.InternalSqlRowImpl;
+import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryPrefetchCallback;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.exec.ExecutablePlan;
@@ -205,6 +206,16 @@ public class KeyValueGetPlan implements ExplainablePlan, ExecutablePlan {
result.whenCompleteAsync((res, err) -> firstPageReadyCallback.onPrefetchComplete(err), executor);
}
+ CompletableFuture<Void> timeoutFut = ctx.timeoutFuture();
+ if (timeoutFut != null) {
+ Executor executor = task -> ctx.execute(task::run, (err) -> {});
+
+ timeoutFut.thenAcceptAsync(
+ (r) -> result.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG)),
+ executor
+ );
+ }
+
return new AsyncWrapper<>(result, Runnable::run);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
index 71b0284fe5..45c2cbddc0 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/KeyValueModifyPlan.java
@@ -27,6 +27,7 @@ import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.SqlExplainLevel;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.InternalSqlRowSingleLong;
+import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryPrefetchCallback;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.exec.ExecutablePlan;
@@ -141,6 +142,16 @@ public class KeyValueModifyPlan implements ExplainablePlan, ExecutablePlan {
result.whenCompleteAsync((res, err) -> firstPageReadyCallback.onPrefetchComplete(err), executor);
}
+ CompletableFuture<Void> timeoutFut = ctx.timeoutFuture();
+ if (timeoutFut != null) {
+ Executor executor = task -> ctx.execute(task::run, (err) -> {});
+
+ timeoutFut.thenAcceptAsync(
+ (r) -> result.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG)),
+ executor
+ );
+ }
+
return new AsyncWrapper<>(result, Runnable::run);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
index 46bbda666f..e7591a7b7b 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PlanningContext.java
@@ -151,7 +151,7 @@ public final class PlanningContext implements Context {
private final long plannerTimeout;
/** Flag indicated if planning has been canceled due to timeout. */
- private boolean timeouted = false;
+ private volatile boolean timeouted;
private final Int2ObjectMap<Object> parameters;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
index e3519f1e2d..d15ceb0201 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImpl.java
@@ -22,7 +22,7 @@ import static org.apache.ignite.internal.sql.engine.prepare.PlannerHelper.optimi
import static org.apache.ignite.internal.sql.engine.trait.TraitUtils.distributionPresent;
import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.thread.ThreadOperation.NOTHING_ALLOWED;
-import static org.apache.ignite.lang.ErrorGroups.Sql.PLANNING_TIMEOUT_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
import java.util.ArrayList;
import java.util.Collections;
@@ -36,7 +36,6 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
-import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
import org.apache.calcite.schema.SchemaPlus;
@@ -53,6 +52,7 @@ import org.apache.ignite.internal.sql.ColumnMetadataImpl;
import org.apache.ignite.internal.sql.ResultSetMetadataImpl;
import org.apache.ignite.internal.sql.configuration.distributed.SqlDistributedConfiguration;
import org.apache.ignite.internal.sql.configuration.local.SqlLocalConfiguration;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
@@ -124,7 +124,7 @@ public class PrepareServiceImpl implements PrepareService {
* @param cacheFactory A factory to create cache of query plans.
* @param dataStorageManager Data storage manager.
* @param metricManager Metric manager.
- * @param clusterCfg Cluster SQL configuration.
+ * @param clusterCfg Cluster SQL configuration.
* @param nodeCfg Node SQL configuration.
* @param schemaManager Schema manager to use on validation phase to bind identifiers in AST with particular schema objects.
*/
@@ -226,6 +226,11 @@ public class PrepareServiceImpl implements PrepareService {
return CompletableFuture.failedFuture(new SchemaNotFoundException(schemaName));
}
+ // Add an action to trigger planner timeout, when operation times out.
+ // Or trigger timeout immediately if operation has already timed out.
+ QueryCancel cancelHandler = operationContext.cancel();
+ assert cancelHandler != null;
+
PlanningContext planningContext = PlanningContext.builder()
.frameworkConfig(Frameworks.newConfigBuilder(FRAMEWORK_CONFIG).defaultSchema(schema).build())
.query(parsedResult.originalQuery())
@@ -237,11 +242,6 @@ public class PrepareServiceImpl implements PrepareService {
return result.exceptionally(ex -> {
Throwable th = ExceptionUtils.unwrapCause(ex);
- if (planningContext.timeouted() && th instanceof RelOptPlanner.CannotPlanException) {
- throw new SqlException(
- PLANNING_TIMEOUT_ERR,
- "Planning of a query aborted due to planner timeout threshold is reached");
- }
throw new CompletionException(SqlExceptionMapperUtil.mapToPublicSqlException(th));
}
@@ -356,25 +356,20 @@ public class PrepareServiceImpl implements PrepareService {
SqlNode validatedNode = validated.sqlNode();
- IgniteRel igniteRel = optimize(validatedNode, planner);
-
- // cluster keeps a lot of cached stuff that won't be used anymore.
- // In order let GC collect that, let's reattach tree to an empty cluster
- // before storing tree in plan cache
- IgniteRel clonedTree = Cloner.clone(igniteRel, Commons.emptyCluster());
+ IgniteRel optimizedRel = doOptimize(ctx, validatedNode, planner, key);
ResultSetMetadata resultSetMetadata = resultSetMetadata(validated.dataType(), validated.origins(), validated.aliases());
int catalogVersion = ctx.catalogVersion();
- if (clonedTree instanceof IgniteKeyValueGet) {
+ if (optimizedRel instanceof IgniteKeyValueGet) {
return new KeyValueGetPlan(
- nextPlanId(), catalogVersion, (IgniteKeyValueGet) clonedTree, resultSetMetadata, parameterMetadata
+ nextPlanId(), catalogVersion, (IgniteKeyValueGet) optimizedRel, resultSetMetadata, parameterMetadata
);
}
var plan = new MultiStepPlan(
- nextPlanId(), SqlQueryType.QUERY, clonedTree, resultSetMetadata, parameterMetadata, catalogVersion
+ nextPlanId(), SqlQueryType.QUERY, optimizedRel, resultSetMetadata, parameterMetadata, catalogVersion
);
if (LOG.isDebugEnabled()) {
@@ -428,24 +423,18 @@ public class PrepareServiceImpl implements PrepareService {
SqlNode validatedNode = stmt.value;
ParameterMetadata parameterMetadata = stmt.parameterMetadata;
- // Convert to Relational operators graph
- IgniteRel igniteRel = optimize(validatedNode, planner);
-
- // cluster keeps a lot of cached stuff that won't be used anymore.
- // In order let GC collect that, let's reattach tree to an empty cluster
- // before storing tree in plan cache
- IgniteRel clonedTree = Cloner.clone(igniteRel, Commons.emptyCluster());
+ IgniteRel optimizedRel = doOptimize(ctx, validatedNode, planner, key);
int catalogVersion = ctx.catalogVersion();
- if (clonedTree instanceof IgniteKeyValueModify) {
+ if (optimizedRel instanceof IgniteKeyValueModify) {
return new KeyValueModifyPlan(
- nextPlanId(), catalogVersion, (IgniteKeyValueModify) clonedTree, DML_METADATA, parameterMetadata
+ nextPlanId(), catalogVersion, (IgniteKeyValueModify) optimizedRel, DML_METADATA, parameterMetadata
);
}
return new MultiStepPlan(
- nextPlanId(), SqlQueryType.DML, clonedTree, DML_METADATA, parameterMetadata, catalogVersion
+ nextPlanId(), SqlQueryType.DML, optimizedRel, DML_METADATA, parameterMetadata, catalogVersion
);
}, planningPool));
@@ -582,6 +571,33 @@ public class PrepareServiceImpl implements PrepareService {
);
}
+ private IgniteRel doOptimize(PlanningContext ctx, SqlNode validatedNode, IgnitePlanner planner, CacheKey key) {
+ // Convert to Relational operators graph
+ IgniteRel igniteRel;
+ try {
+ igniteRel = optimize(validatedNode, planner);
+ } catch (Exception e) {
+ // Remove this cache entry if planning timed out.
+ // Otherwise the cache will keep a plan that can not be used anymore,
+ // and we should allow another planning attempt with increased timeout.
+ if (ctx.timeouted()) {
+ cache.invalidate(key);
+
+ //noinspection ThrowInsideCatchBlockWhichIgnoresCaughtException
+ throw new SqlException(
+ EXECUTION_CANCELLED_ERR,
+ "Planning of a query aborted due to planner timeout threshold is reached");
+ } else {
+ throw new CompletionException(e);
+ }
+ }
+
+ // cluster keeps a lot of cached stuff that won't be used anymore.
+ // In order let GC collect that, let's reattach tree to an empty cluster
+ // before storing tree in plan cache
+ return Cloner.clone(igniteRel, Commons.emptyCluster());
+ }
+
private static ParameterMetadata createParameterMetadata(RelDataType parameterRowType) {
if (parameterRowType.getFieldCount() == 0) {
return EMPTY_PARAMETER_METADATA;
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/Cache.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/Cache.java
index 92e783d3e5..c56ac52a10 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/Cache.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/Cache.java
@@ -79,4 +79,18 @@ public interface Cache<K, V> {
* @param valueFilter A predicate which returns {@code true} for the values of entries to be removed.
*/
void removeIfValue(Predicate<? super V> valueFilter);
+
+ /**
+ * Invalidates a cache entry for the given key.
+ *
+ * @param key Key.
+ */
+ void invalidate(K key);
+
+ /**
+ * Returns the number of entries in the cache.
+ *
+ * @return The number of entries in the cache.
+ */
+ int size();
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
index 45c597ce64..4b4c01824c 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/util/cache/CaffeineCacheFactory.java
@@ -148,5 +148,15 @@ public class CaffeineCacheFactory implements CacheFactory {
public void removeIfValue(Predicate<? super V> valueFilter) {
cache.asMap().values().removeIf(valueFilter);
}
+
+ @Override
+ public void invalidate(K key) {
+ cache.invalidate(key);
+ }
+
+ @Override
+ public int size() {
+ return cache.asMap().size();
+ }
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index feac21c78a..912c98cf97 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -41,10 +41,14 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -59,13 +63,16 @@ import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.ignite.internal.catalog.CatalogCommand;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalNode;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologySnapshot;
import org.apache.ignite.internal.failure.FailureProcessor;
@@ -114,10 +121,14 @@ import org.apache.ignite.internal.sql.engine.message.MessageService;
import org.apache.ignite.internal.sql.engine.message.QueryStartRequest;
import org.apache.ignite.internal.sql.engine.message.QueryStartResponseImpl;
import org.apache.ignite.internal.sql.engine.message.SqlQueryMessagesFactory;
+import org.apache.ignite.internal.sql.engine.prepare.DdlPlan;
+import org.apache.ignite.internal.sql.engine.prepare.KeyValueGetPlan;
+import org.apache.ignite.internal.sql.engine.prepare.KeyValueModifyPlan;
import org.apache.ignite.internal.sql.engine.prepare.PrepareService;
import org.apache.ignite.internal.sql.engine.prepare.PrepareServiceImpl;
import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
-import org.apache.ignite.internal.sql.engine.prepare.pruning.PartitionPrunerImpl;
+import org.apache.ignite.internal.sql.engine.prepare.ddl.DdlSqlToCommandConverter;
+import org.apache.ignite.internal.sql.engine.prepare.pruning.PartitionPruner;
import org.apache.ignite.internal.sql.engine.rel.IgniteTableScan;
import org.apache.ignite.internal.sql.engine.schema.IgniteSchema;
import org.apache.ignite.internal.sql.engine.schema.IgniteSystemView;
@@ -129,6 +140,7 @@ import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.util.EmptyCacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.type.NativeTypes;
import org.apache.ignite.internal.util.AsyncCursor;
import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
@@ -169,9 +181,10 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
private final IgniteTable table = TestBuilders.table()
.name("TEST_TBL")
- .addColumn("ID", NativeTypes.INT32)
+ .addKeyColumn("ID", NativeTypes.INT32)
.addColumn("VAL", NativeTypes.INT32)
- .distribution(IgniteDistributions.random())
+ .distribution(IgniteDistributions.affinity(0, 1, 2))
+ .hashIndex().name("TEST_TBL_PK").addColumn("ID").primaryKey(true).end()
.size(1_000_000)
.build();
@@ -187,6 +200,8 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
private final List<QueryTaskExecutor> executers = new ArrayList<>();
+ private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
private ClusterNode firstNode;
@BeforeEach
@@ -197,7 +212,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
"test",
0,
CaffeineCacheFactory.INSTANCE,
- null,
+ new DdlSqlToCommandConverter(),
PLANNING_TIMEOUT,
PLANNING_THREAD_COUNT,
new MetricManagerImpl(),
@@ -227,6 +242,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
});
executers.clear();
+ scheduler.shutdownNow();
}
/**
@@ -756,9 +772,30 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
assertThat(cursor.closeAsync(), willCompleteSuccessfully());
}
+ @Test
+ public void testExecuteCancelled() {
+ ExecutionService execService = executionServices.get(0);
+
+ QueryCancel cancel = new QueryCancel();
+ SqlOperationContext ctx = operationContext(null)
+ .cancel(cancel)
+ .build();
+
+ QueryPlan plan = prepare("SELECT * FROM test_tbl", ctx);
+
+ // Cancel the query
+ cancel.cancel();
+
+ // Should immediately trigger query cancel exception.
+ IgniteTestUtils.assertThrows(QueryCancelledException.class,
+ () -> execService.executePlan(plan, ctx),
+ "The query was cancelled while executing"
+ );
+ }
+
/**
- * Test checks the format of the debugging information dump obtained during query execution.
- * To obtain verifiable results, all response messages are blocked while debugging information is obtained.
+ * Test checks the format of the debugging information dump obtained during query execution. To obtain verifiable results, all response
+ * messages are blocked while debugging information is obtained.
*/
@Test
public void testDebugInfoFormat() throws InterruptedException {
@@ -837,6 +874,242 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
assertThat(debugInfo3, equalTo(expectedOnNonCoordinator));
}
+ @Test
+ public void testTimeoutStandardPlan() {
+ nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
+
+ int deadlineMillis = 500;
+
+ testCluster.node(nodeNames.get(2)).interceptor((nodeName, msg, original) -> {
+ if (msg instanceof QueryStartRequest) {
+ CompletableFuture<Void> f = new CompletableFuture<>();
+ f.thenRun(() -> original.onMessage(nodeName, msg));
+ f.completeOnTimeout(null, deadlineMillis * 2, TimeUnit.MILLISECONDS);
+ } else {
+ original.onMessage(nodeName, msg);
+ }
+
+ return nullCompletedFuture();
+ });
+
+ ExecutionService execService = executionServices.get(0);
+
+ // Use a separate context, so planning won't timeout.
+ SqlOperationContext planCtx = operationContext(null).build();
+ QueryPlan plan = prepare("SELECT * FROM test_tbl", planCtx);
+
+ int attempts = 10;
+
+ for (int k = 0; k < attempts; k++) {
+ QueryCancel cancel = new QueryCancel();
+ CompletableFuture<Void> timeoutFut = cancel.setTimeout(scheduler, deadlineMillis);
+
+ SqlOperationContext execCtx = operationContext(null)
+ .cancel(cancel)
+ .build();
+
+ AsyncCursor<InternalSqlRow> cursor;
+ try {
+ cursor = execService.executePlan(plan, execCtx);
+ } catch (QueryCancelledException e) {
+ // This might happen when initialization took longer than a time out,
+ // Retry to get a proper error.
+ continue;
+ }
+
+ CompletableFuture<?> batchFut = cursor.requestNextAsync(1);
+
+ timeoutFut.join();
+
+ IgniteTestUtils.assertThrowsWithCause(
+ batchFut::join,
+ SqlException.class,
+ "Query timeout"
+ );
+
+ return;
+ }
+
+ fail("Failed to get query timeout error");
+ }
+
+ @Test
+ public void testTimeoutEarly() {
+ ExecutionService execService = executionServices.get(0);
+
+ // Use a separate context, so planning won't timeout.
+ SqlOperationContext planCtx = operationContext(null).build();
+ QueryPlan plan = prepare("SELECT * FROM test_tbl", planCtx);
+
+ QueryCancel queryCancel = new QueryCancel();
+ CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, 0);
+
+ SqlOperationContext ctx = operationContext(null)
+ .cancel(queryCancel)
+ .build();
+
+ // Cancel the query with a timeout
+ timeoutFut.join();
+
+ // Should immediately trigger query cancel exception.
+ IgniteTestUtils.assertThrows(QueryCancelledException.class,
+ () -> execService.executePlan(plan, ctx),
+ "Query timeout"
+ );
+ }
+
+ @Test
+ public void testTimeoutKvGet() {
+ int deadlineMillis = 500;
+
+ // Use a separate context, so planning won't timeout.
+ SqlOperationContext planCtx = operationContext(null).build();
+ QueryPlan plan = prepare("SELECT * FROM test_tbl WHERE id = 1", planCtx);
+
+ assertInstanceOf(KeyValueGetPlan.class, plan);
+
+ ExecutionServiceImpl<?> execService = executionServices.get(0);
+ NoOpExecutableTableRegistry tableRegistry = (NoOpExecutableTableRegistry) execService.tableRegistry();
+
+ Duration delay = Duration.of(deadlineMillis * 2, ChronoUnit.MILLIS);
+ tableRegistry.setGetTableDelay(delay);
+
+ int attempts = 10;
+
+ for (int k = 0; k < attempts; k++) {
+ QueryCancel queryCancel = new QueryCancel();
+ CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, deadlineMillis);
+
+ SqlOperationContext execCtx = operationContext(null)
+ .cancel(queryCancel)
+ .build();
+
+ AsyncCursor<InternalSqlRow> cursor;
+ try {
+ cursor = execService.executePlan(plan, execCtx);
+ } catch (QueryCancelledException e) {
+ continue;
+ }
+
+ CompletableFuture<?> batchFut = cursor.requestNextAsync(1);
+
+ timeoutFut.join();
+
+ IgniteTestUtils.assertThrowsWithCause(
+ batchFut::join,
+ SqlException.class,
+ "Query timeout"
+ );
+
+ return;
+ }
+
+ fail("Failed to get query timeout error");
+ }
+
+ @Test
+ public void testTimeoutKvModify() {
+ // Use a separate context, so planning won't timeout.
+ SqlOperationContext planCtx = operationContext(null).build();
+ QueryPlan plan = prepare("INSERT INTO test_tbl VALUES(1, 2)", planCtx);
+
+ assertInstanceOf(KeyValueModifyPlan.class, plan);
+
+ int deadlineMillis = 500;
+
+ ExecutionServiceImpl<?> execService = executionServices.get(0);
+ NoOpExecutableTableRegistry tableRegistry = (NoOpExecutableTableRegistry) execService.tableRegistry();
+
+ Duration delay = Duration.of(deadlineMillis * 2, ChronoUnit.MILLIS);
+ tableRegistry.setGetTableDelay(delay);
+
+ int attempts = 10;
+
+ for (int k = 0; k < attempts; k++) {
+ QueryCancel queryCancel = new QueryCancel();
+ CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, deadlineMillis);
+
+ SqlOperationContext execCtx = operationContext(null)
+ .cancel(queryCancel)
+ .build();
+
+ AsyncCursor<InternalSqlRow> cursor;
+ try {
+ cursor = execService.executePlan(plan, execCtx);
+ } catch (QueryCancelledException e) {
+ continue;
+ }
+
+ CompletableFuture<?> batchFut = cursor.requestNextAsync(1);
+
+ timeoutFut.join();
+
+ IgniteTestUtils.assertThrowsWithCause(
+ batchFut::join,
+ SqlException.class,
+ "Query timeout"
+ );
+
+ return;
+ }
+
+ fail("Failed to get query timeout error");
+ }
+
+ @Test
+ public void testTimeoutDdl() {
+ // Use a separate context, so planning won't timeout.
+ SqlOperationContext planCtx = operationContext(null).build();
+ QueryPlan plan = prepare("CREATE TABLE x (id INTEGER PRIMARY KEY, val INTEGER)", planCtx);
+
+ assertInstanceOf(DdlPlan.class, plan);
+
+ int deadlineMillis = 500;
+
+ ExecutionServiceImpl<?> execService = executionServices.get(0);
+
+ NoOpExecutableTableRegistry tableRegistry = (NoOpExecutableTableRegistry) execService.tableRegistry();
+
+ Duration delay = Duration.of(deadlineMillis * 2, ChronoUnit.MILLIS);
+ tableRegistry.setGetTableDelay(delay);
+
+ DdlCommandHandler ddlCommandHandler = execService.ddlCommandHandler();
+ when(ddlCommandHandler.handle(any(CatalogCommand.class))).thenReturn(new CompletableFuture<>());
+
+ int attempts = 10;
+
+ for (int k = 0; k < attempts; k++) {
+ QueryCancel queryCancel = new QueryCancel();
+ CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, deadlineMillis);
+
+ SqlOperationContext execCtx = operationContext(null)
+ .cancel(queryCancel)
+ .build();
+
+ AsyncCursor<InternalSqlRow> cursor;
+ try {
+ cursor = execService.executePlan(plan, execCtx);
+ } catch (QueryCancelledException e) {
+ continue;
+ }
+
+ CompletableFuture<?> batchFut = cursor.requestNextAsync(1);
+
+ timeoutFut.join();
+
+ // DDL handler does convert exceptions to SqlException, so we get QueryCancelledException here.
+ IgniteTestUtils.assertThrowsWithCause(
+ batchFut::join,
+ QueryCancelledException.class,
+ "Query timeout"
+ );
+
+ return;
+ }
+
+ fail("Failed to get query timeout error");
+ }
+
/** Creates an execution service instance for the node with given consistent id. */
public ExecutionServiceImpl<Object[]> create(String nodeName) {
if (!nodeNames.contains(nodeName)) {
@@ -931,7 +1204,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
}
};
- var partitionPruner = new PartitionPrunerImpl();
+ PartitionPruner partitionPruner = (mappedFragments, dynamicParameters) -> mappedFragments;
return new MappingServiceImpl(nodeName, clock, targetProvider, EmptyCacheFactory.INSTANCE, 0, partitionPruner, taskExecutor);
}
@@ -940,6 +1213,10 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
}
private SqlOperationContext createContext(@Nullable PrefetchCallback prefetchCallback) {
+ return operationContext(prefetchCallback).build();
+ }
+
+ private SqlOperationContext.Builder operationContext(@Nullable PrefetchCallback prefetchCallback) {
return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
.cancel(new QueryCancel())
@@ -947,8 +1224,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
.operationTime(new HybridClockImpl().now())
.defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.timeZoneId(SqlQueryProcessor.DEFAULT_TIME_ZONE_ID)
- .txContext(ExplicitTxContext.fromTx(new NoOpTransaction(nodeNames.get(0))))
- .build();
+ .txContext(ExplicitTxContext.fromTx(new NoOpTransaction(nodeNames.get(0))));
}
private QueryPlan prepare(String query, SqlOperationContext ctx) {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
index a06fbcb68a..15e472a1f5 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/NoOpExecutableTableRegistry.java
@@ -17,17 +17,42 @@
package org.apache.ignite.internal.sql.engine.exec;
+import java.time.Duration;
+import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.schema.PartitionCalculator;
import org.apache.ignite.internal.sql.engine.schema.TableDescriptor;
/** Stub implementation for {@link ExecutableTableRegistry}. */
public final class NoOpExecutableTableRegistry implements ExecutableTableRegistry {
+
+ private static final IgniteLogger LOG = Loggers.forClass(NoOpExecutableTableRegistry.class);
+
+ private volatile Duration delay = Duration.ZERO;
+
+ /** Sets a delay for {@link #getTable(int, int)} operation. */
+ public void setGetTableDelay(Duration delay) {
+ this.delay = Objects.requireNonNull(delay, "delay");
+ }
+
/** {@inheritDoc} */
@Override
public CompletableFuture<ExecutableTable> getTable(int catalogVersion, int tableId) {
- return CompletableFuture.completedFuture(new NoOpExecutableTable(tableId));
+ Duration delay = this.delay;
+
+ CompletableFuture<ExecutableTable> f = new CompletableFuture<>();
+
+ LOG.info("Requested tableId={} in catalogVersion={} with delay={}", tableId, catalogVersion, delay);
+
+ f.completeOnTimeout(new NoOpExecutableTable(tableId), delay.toMillis(), TimeUnit.MILLISECONDS).thenRun(() -> {
+ LOG.info("Return tableId={} in catalogVersion={} after delay={}", tableId, catalogVersion, delay);
+ });
+
+ return f;
}
private static final class NoOpExecutableTable implements ExecutableTable {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/RuntimeSortedIndexTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/RuntimeSortedIndexTest.java
index d9ad4dc699..fe8f01d5e4 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/RuntimeSortedIndexTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/RuntimeSortedIndexTest.java
@@ -122,7 +122,8 @@ public class RuntimeSortedIndexTest extends IgniteAbstractTest {
ArrayRowHandler.INSTANCE,
Map.of(),
null,
- SqlQueryProcessor.DEFAULT_TIME_ZONE_ID
+ SqlQueryProcessor.DEFAULT_TIME_ZONE_ID,
+ null
),
RelCollations.of(ImmutableIntList.copyOf(idxCols)),
(o1, o2) -> {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractExecutionTest.java
index d97dff0a05..25abdef0ef 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/AbstractExecutionTest.java
@@ -125,7 +125,8 @@ public abstract class AbstractExecutionTest<T> extends IgniteAbstractTest {
rowHandler(),
Map.of(),
TxAttributes.fromTx(new NoOpTransaction("fake-test-node")),
- SqlQueryProcessor.DEFAULT_TIME_ZONE_ID
+ SqlQueryProcessor.DEFAULT_TIME_ZONE_ID,
+ null
);
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinExecutionTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinExecutionTest.java
index 38d5e52a0a..fbdc13adb0 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinExecutionTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/MergeJoinExecutionTest.java
@@ -486,7 +486,7 @@ public class MergeJoinExecutionTest extends AbstractExecutionTest<Object[]> {
ExecutionContext<Object[]> ectx =
new ExecutionContext<>(null, null, null, null, null,
- ArrayRowHandler.INSTANCE, null, null, SqlQueryProcessor.DEFAULT_TIME_ZONE_ID);
+ ArrayRowHandler.INSTANCE, null, null, SqlQueryProcessor.DEFAULT_TIME_ZONE_ID, null);
ExpressionFactoryImpl<Object[]> expFactory = new ExpressionFactoryImpl<>(ectx, SqlConformanceEnum.DEFAULT);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
index c3a921a731..dba14ca6f4 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestBuilders.java
@@ -412,6 +412,9 @@ public class TestBuilders {
* @see TestIndex
*/
public interface SortedIndexBuilder extends SortedIndexBuilderBase<SortedIndexBuilder>, NestedBuilder<TableBuilder> {
+
+ /** Specifies whether this index is a primary key index or not. */
+ SortedIndexBuilder primaryKey(boolean value);
}
/**
@@ -420,6 +423,9 @@ public class TestBuilders {
* @see TestIndex
*/
public interface HashIndexBuilder extends HashIndexBuilderBase<HashIndexBuilder>, NestedBuilder<TableBuilder> {
+
+ /** Specifies whether this index is a primary key index or not. */
+ HashIndexBuilder primaryKey(boolean value);
}
/**
@@ -552,7 +558,8 @@ public class TestBuilders {
ArrayRowHandler.INSTANCE,
Commons.parametersMap(dynamicParams),
TxAttributes.fromTx(new NoOpTransaction(node.name())),
- SqlQueryProcessor.DEFAULT_TIME_ZONE_ID
+ SqlQueryProcessor.DEFAULT_TIME_ZONE_ID,
+ null
);
}
}
@@ -1062,6 +1069,8 @@ public class TestBuilders {
implements SortedIndexBuilder {
private final TableBuilderImpl parent;
+ private boolean primary;
+
private SortedIndexBuilderImpl(TableBuilderImpl parent) {
this.parent = parent;
}
@@ -1080,6 +1089,13 @@ public class TestBuilders {
return parent;
}
+ /** {@inheritDoc} */
+ @Override
+ public SortedIndexBuilder primaryKey(boolean value) {
+ this.primary = value;
+ return self();
+ }
+
/** {@inheritDoc} */
@Override
public TestIndex build(TableDescriptor desc) {
@@ -1095,13 +1111,15 @@ public class TestBuilders {
throw new IllegalArgumentException("Collation must be specified for each of columns.");
}
- return TestIndex.createSorted(name, columns, collations, desc);
+ return TestIndex.createSorted(name, columns, collations, desc, primary);
}
}
private static class HashIndexBuilderImpl extends AbstractTableIndexBuilderImpl<HashIndexBuilder> implements HashIndexBuilder {
private final TableBuilderImpl parent;
+ private boolean primary;
+
private HashIndexBuilderImpl(TableBuilderImpl parent) {
this.parent = parent;
}
@@ -1120,6 +1138,13 @@ public class TestBuilders {
return parent;
}
+ /** {@inheritDoc} */
+ @Override
+ public HashIndexBuilder primaryKey(boolean value) {
+ this.primary = value;
+ return self();
+ }
+
/** {@inheritDoc} */
@Override
public TestIndex build(TableDescriptor desc) {
@@ -1133,7 +1158,7 @@ public class TestBuilders {
assert collations == null : "Collation is not supported.";
- return TestIndex.createHash(name, columns, desc);
+ return TestIndex.createHash(name, columns, desc, primary);
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
index 188f612a9c..c05ca3c42a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestIndex.java
@@ -33,11 +33,12 @@ public class TestIndex extends IgniteIndex {
static TestIndex createHash(
String name,
List<String> columns,
- TableDescriptor tableDescriptor
+ TableDescriptor tableDescriptor,
+ boolean primary
) {
RelCollation collation = TraitUtils.createCollation(columns, null, tableDescriptor);
- return new TestIndex(name, Type.HASH, tableDescriptor.distribution(), collation);
+ return new TestIndex(name, Type.HASH, tableDescriptor.distribution(), collation, primary);
}
/** Factory method for creating sorted-index. */
@@ -45,11 +46,12 @@ public class TestIndex extends IgniteIndex {
String name,
List<String> columns,
List<Collation> collations,
- TableDescriptor tableDescriptor
+ TableDescriptor tableDescriptor,
+ boolean primary
) {
RelCollation collation = TraitUtils.createCollation(columns, collations, tableDescriptor);
- return new TestIndex(name, Type.SORTED, tableDescriptor.distribution(), collation);
+ return new TestIndex(name, Type.SORTED, tableDescriptor.distribution(), collation, primary);
}
private static final AtomicInteger ID = new AtomicInteger();
@@ -59,8 +61,9 @@ public class TestIndex extends IgniteIndex {
String name,
Type type,
IgniteDistribution distribution,
- RelCollation collation
+ RelCollation collation,
+ boolean primary
) {
- super(ID.incrementAndGet(), name, type, distribution, collation);
+ super(ID.incrementAndGet(), name, type, distribution, collation, primary);
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
index 1f7482fe1a..3403eab1d0 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/AbstractPlannerTest.java
@@ -79,6 +79,7 @@ import org.apache.calcite.util.Util;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.lang.IgniteStringBuilder;
import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.exec.mapping.IdGenerator;
import org.apache.ignite.internal.sql.engine.exec.mapping.QuerySplitter;
@@ -306,6 +307,7 @@ public abstract class AbstractPlannerTest extends IgniteAbstractTest {
.timeZoneId(ZoneId.systemDefault())
.operationTime(new HybridClockImpl().now())
.defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
+ .cancel(new QueryCancel())
.parameters(params)
.build();
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
index 2fb6cd2d56..c7b51271fe 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/planner/PlannerTimeoutTest.java
@@ -21,7 +21,7 @@ import static org.apache.calcite.tools.Frameworks.newConfigBuilder;
import static org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
-import static org.apache.ignite.lang.ErrorGroups.Sql.PLANNING_TIMEOUT_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -76,7 +76,7 @@ public class PlannerTimeoutTest extends AbstractPlannerTest {
ParsedResult parsedResult = parserService.parse("SELECT * FROM T1 t, T1 t1, T1 t2, T1 t3");
SqlTestUtils.assertThrowsSqlException(
- PLANNING_TIMEOUT_ERR,
+ EXECUTION_CANCELLED_ERR,
"Planning of a query aborted due to planner timeout threshold is reached",
() -> await(prepareService.prepareAsync(parsedResult, ctx)));
} finally {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
index bc410aa408..5374e89f7e 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/prepare/PrepareServiceImplTest.java
@@ -18,25 +18,31 @@
package org.apache.ignite.internal.sql.engine.prepare;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.assertThrowsWithCause;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.sameInstance;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.metrics.MetricManagerImpl;
import org.apache.ignite.internal.sql.SqlCommon;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.SqlOperationContext;
import org.apache.ignite.internal.sql.engine.framework.PredefinedSchemaManager;
import org.apache.ignite.internal.sql.engine.framework.TestBuilders;
@@ -47,18 +53,25 @@ import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.trait.IgniteDistributions;
import org.apache.ignite.internal.sql.engine.util.SqlTestUtils;
+import org.apache.ignite.internal.sql.engine.util.cache.Cache;
+import org.apache.ignite.internal.sql.engine.util.cache.CacheFactory;
import org.apache.ignite.internal.sql.engine.util.cache.CaffeineCacheFactory;
+import org.apache.ignite.internal.sql.engine.util.cache.StatsCounter;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
import org.apache.ignite.internal.type.NativeType;
import org.apache.ignite.internal.type.NativeTypes;
+import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.sql.ColumnMetadata;
import org.apache.ignite.sql.ColumnType;
+import org.apache.ignite.sql.SqlException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.Mockito;
/**
* Tests to verify {@link PrepareServiceImpl}.
@@ -66,6 +79,8 @@ import org.junit.jupiter.params.provider.MethodSource;
public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
private static final List<PrepareService> createdServices = new ArrayList<>();
+ private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
+
@AfterEach
public void stopServices() throws Exception {
for (PrepareService createdService : createdServices) {
@@ -75,6 +90,11 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
createdServices.clear();
}
+ @AfterEach
+ public void stopScheduler() {
+ scheduler.shutdownNow();
+ }
+
@Test
public void prepareServiceReturnsExistingPlanForExplain() {
PrepareService service = createPlannerService();
@@ -222,6 +242,73 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
assertTrue(parameterType.nullable(), "Nullabilty does not match: " + parameterType);
}
+ @Test
+ public void timeoutedPlanShouldBeRemovedFromCache() throws InterruptedException {
+ IgniteTable igniteTable = TestBuilders.table()
+ .name("T")
+ .addColumn("C", NativeTypes.INT32)
+ .distribution(IgniteDistributions.single())
+ .build();
+
+ // Create a proxy.
+ IgniteTable spyTable = Mockito.spy(igniteTable);
+
+ // Override and slowdown a method, which is called by Planner, to emulate long planning.
+ Mockito.doAnswer(inv -> {
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ // Call original method.
+ return igniteTable.getRowType(inv.getArgument(0), inv.getArgument(1));
+ }).when(spyTable).getRowType(any(), any());
+
+ IgniteSchema schema = new IgniteSchema("PUBLIC", 0, List.of(igniteTable));
+ Cache<Object, Object> cache = CaffeineCacheFactory.INSTANCE.create(100);
+
+ CacheFactory cacheFactory = new CacheFactory() {
+ @Override
+ public <K, V> Cache<K, V> create(int size) {
+ return (Cache<K, V>) cache;
+ }
+
+ @Override
+ public <K, V> Cache<K, V> create(int size, StatsCounter statCounter) {
+ return (Cache<K, V>) cache;
+ }
+ };
+
+ PrepareServiceImpl service = createPlannerService(schema, cacheFactory, 100);
+
+ StringBuilder stmt = new StringBuilder();
+ for (int i = 0; i < 100; i++) {
+ if (i > 0) {
+ stmt.append("UNION");
+ stmt.append(System.lineSeparator());
+ }
+ stmt.append("SELECT * FROM t WHERE c = ").append(i);
+ stmt.append(System.lineSeparator());
+ }
+
+ ParsedResult parsedResult = parse(stmt.toString());
+
+ SqlOperationContext context = operationContext().build();
+
+ Throwable err = assertThrowsWithCause(
+ () -> service.prepareAsync(parsedResult, context).get(),
+ SqlException.class
+ );
+
+ Throwable cause = ExceptionUtils.unwrapCause(err);
+ SqlException sqlErr = assertInstanceOf(SqlException.class, cause, "Unexpected error. Root error: " + err);
+ assertEquals(Sql.EXECUTION_CANCELLED_ERR, sqlErr.code(), "Unexpected error: " + sqlErr);
+
+ // Cache invalidate does not immediately remove the entry, so we need to wait some time to ensure it is removed.
+ boolean empty = IgniteTestUtils.waitForCondition(() -> cache.size() == 0, 1000);
+ assertTrue(empty, "Cache is not empty: " + cache.size());
+ }
+
private static Stream<Arguments> parameterTypes() {
int noScale = ColumnMetadata.UNDEFINED_SCALE;
int noPrecision = ColumnMetadata.UNDEFINED_PRECISION;
@@ -250,13 +337,17 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
}
private static SqlOperationContext createContext(Object... params) {
+ return operationContext(params).build();
+ }
+
+ private static SqlOperationContext.Builder operationContext(Object... params) {
return SqlOperationContext.builder()
.queryId(UUID.randomUUID())
.timeZoneId(ZoneId.systemDefault())
.operationTime(new HybridClockImpl().now())
.defaultSchemaName(SqlCommon.DEFAULT_SCHEMA_NAME)
.parameters(params)
- .build();
+ .cancel(new QueryCancel());
}
private static IgniteSchema createSchema() {
@@ -275,8 +366,12 @@ public class PrepareServiceImplTest extends BaseIgniteAbstractTest {
}
private static PrepareService createPlannerService(IgniteSchema schema) {
- PrepareService service = new PrepareServiceImpl("test", 1_000, CaffeineCacheFactory.INSTANCE,
- mock(DdlSqlToCommandConverter.class), 5_000, 2, mock(MetricManagerImpl.class),
+ return createPlannerService(schema, CaffeineCacheFactory.INSTANCE, 1000);
+ }
+
+ private static PrepareServiceImpl createPlannerService(IgniteSchema schema, CacheFactory cacheFactory, int timeoutMillis) {
+ PrepareServiceImpl service = new PrepareServiceImpl("test", 1000, cacheFactory,
+ mock(DdlSqlToCommandConverter.class), timeoutMillis, 2, mock(MetricManagerImpl.class),
new PredefinedSchemaManager(schema));
createdServices.add(service);
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/util/EmptyCacheFactory.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/util/EmptyCacheFactory.java
index 12623d8ce2..d966b19f85 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/util/EmptyCacheFactory.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/util/EmptyCacheFactory.java
@@ -80,5 +80,15 @@ public class EmptyCacheFactory implements CacheFactory {
public void removeIfValue(Predicate<? super V> valueFilter) {
// NO-OP.
}
+
+ @Override
+ public void invalidate(K key) {
+ // NO-OP.
+ }
+
+ @Override
+ public int size() {
+ return 0;
+ }
}
}
diff --git a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
index 15ac31740f..f1ef02ce0c 100644
--- a/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
+++ b/modules/sql-engine/src/testFixtures/java/org/apache/ignite/internal/sql/engine/util/QueryCheckerImpl.java
@@ -297,11 +297,13 @@ abstract class QueryCheckerImpl implements QueryChecker {
// Check plan.
QueryProcessor qryProc = getEngine();
- SqlProperties properties = SqlPropertiesHelper.newBuilder()
+ SqlProperties newProperties = SqlPropertiesHelper.newBuilder()
.set(QueryProperty.ALLOWED_QUERY_TYPES, SqlQueryType.SINGLE_STMT_TYPES)
.set(QueryProperty.TIME_ZONE_ID, timeZoneId)
.build();
+ SqlProperties properties = SqlPropertiesHelper.merge(newProperties, SqlQueryProcessor.DEFAULT_PROPERTIES);
+
String qry = queryTemplate.createQuery();
LOG.info("Executing query: [nodeName={}, query={}]", nodeName(), qry);