You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ko...@apache.org on 2024/11/04 07:32:52 UTC
(ignite-3) branch main updated: IGNITE-23544 Sql. Rework query execution pipeline (#4653)
This is an automated email from the ASF dual-hosted git repository.
korlov 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 f1b1b6061f IGNITE-23544 Sql. Rework query execution pipeline (#4653)
f1b1b6061f is described below
commit f1b1b6061f26057631fd2703d3421ec534d5402b
Author: korlov42 <ko...@apache.org>
AuthorDate: Mon Nov 4 09:32:47 2024 +0200
IGNITE-23544 Sql. Rework query execution pipeline (#4653)
---
.../ignite/internal/util/SubscriptionUtils.java | 28 +-
.../subscription/IterableToPublisherAdapter.java | 33 +-
.../sql/engine/ItSqlQueriesSystemViewTest.java | 12 +-
...nPhase.java => InflightTransactionTracker.java} | 35 +-
.../ignite/internal/sql/engine/QueryCancel.java | 123 +---
.../internal/sql/engine/SqlOperationContext.java | 27 +-
.../sql/engine/SqlQueriesViewProvider.java | 109 ++++
.../internal/sql/engine/SqlQueryProcessor.java | 678 ++-------------------
.../internal/sql/engine/exec/ExecutionContext.java | 23 +-
.../sql/engine/exec/ExecutionServiceImpl.java | 40 +-
.../TransactionTracker.java} | 25 +-
.../exec/fsm/CursorInitializationPhaseHandler.java | 109 ++++
.../sql/engine/exec/fsm/ExecutionPhase.java | 73 +++
.../fsm/ExecutionPhaseHandler.java} | 24 +-
.../sql/engine/exec/fsm/MultiStatementHandler.java | 281 +++++++++
.../engine/exec/fsm/OptimizingPhaseHandler.java | 107 ++++
.../sql/engine/exec/fsm/ParsingPhaseHandler.java | 86 +++
.../ignite/internal/sql/engine/exec/fsm/Query.java | 193 ++++++
.../sql/engine/exec/fsm/QueryExecutor.java | 342 +++++++++++
.../internal/sql/engine/exec/fsm/QueryInfo.java | 114 ++++
.../fsm/RegisteredPhaseHandler.java} | 28 +-
.../internal/sql/engine/exec/fsm/Result.java | 67 ++
.../exec/fsm/ScriptInitializationPhaseHandler.java | 47 ++
.../fsm/StopHandler.java} | 26 +-
.../sql/engine/exec/fsm/ValidationHelper.java | 76 +++
.../engine/registry/RunningQueriesRegistry.java | 52 --
.../registry/RunningQueriesRegistryImpl.java | 142 -----
.../sql/engine/registry/RunningQueryInfo.java | 74 ---
.../sql/engine/registry/RunningQueryInfoImpl.java | 148 -----
.../engine/registry/RunningScriptInfoTracker.java | 45 --
.../registry/RunningScriptInfoTrackerImpl.java | 103 ----
.../sql/engine/tx/QueryTransactionContextImpl.java | 14 +-
.../sql/engine/tx/QueryTransactionWrapperImpl.java | 12 +-
.../sql/engine/tx/ScriptTransactionContext.java | 10 +-
.../engine/tx/ScriptTransactionWrapperImpl.java | 10 +-
.../engine/QueryTransactionWrapperSelfTest.java | 40 +-
.../sql/engine/exec/ExecutionServiceImplTest.java | 19 +-
.../sql/engine/exec/TxAwareCursorSelfTest.java | 3 +-
.../sql/engine/framework/ExplicitTxContext.java | 16 +-
.../sql/engine/framework/ImplicitTxContext.java | 16 +-
.../engine/framework/NoOpTransactionTracker.java} | 30 +-
.../sql/engine/framework/NoopRunningQueryInfo.java | 113 ----
.../internal/sql/engine/framework/TestNode.java | 1 -
.../registry/RunningQueriesRegistryTest.java | 224 -------
44 files changed, 1917 insertions(+), 1861 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/SubscriptionUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/util/SubscriptionUtils.java
index 41b4f81d6b..76fab9b412 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/SubscriptionUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/SubscriptionUtils.java
@@ -22,7 +22,9 @@ import java.util.Arrays;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Flow.Publisher;
+import java.util.concurrent.Flow.Subscriber;
import org.apache.ignite.internal.util.subscription.ConcatenatedPublisher;
import org.apache.ignite.internal.util.subscription.IterableToPublisherAdapter;
import org.apache.ignite.internal.util.subscription.OrderedMergePublisher;
@@ -60,7 +62,7 @@ public class SubscriptionUtils {
* @param comparator Rows comparator.
* @param prefetch Prefetch size.
* @param source Iterator of upstream publishers. Each of passed publishers should have guarantee to be sorted with the same
- * comparator.
+ * comparator.
* @return The publisher will combine all of the passed sources into a single one with sorting guaranties.
*/
public static <T> Publisher<T> orderedMerge(Comparator<T> comparator, int prefetch, Iterator<Publisher<? extends T>> source) {
@@ -75,7 +77,8 @@ public class SubscriptionUtils {
*
* @param comparator Rows comparator.
* @param prefetch Prefetch size.
- * @param sources Array of upstream publishers. Each of passed publishers should have guarantee to be sorted with the same comparator.
+ * @param sources Array of upstream publishers. Each of passed publishers should have guarantee to be sorted with the same
+ * comparator.
* @return The publisher will combine all of the passed sources into a single one with sorting guaranties.
*/
@SafeVarargs
@@ -96,6 +99,25 @@ public class SubscriptionUtils {
* @return Publisher created from the given iterable.
*/
public static <T> Publisher<T> fromIterable(Iterable<T> iterable) {
- return new IterableToPublisherAdapter<>(iterable, Runnable::run, Integer.MAX_VALUE);
+ return new IterableToPublisherAdapter<>(CompletableFuture.completedFuture(iterable), Runnable::run, Integer.MAX_VALUE);
+ }
+
+ /**
+ * Creates a publisher from the given iterable.
+ *
+ * <p>A new iterator will be issued for every new subscription.
+ *
+ * <p>Created publisher will start to emit items when future enclosing iterable will be completed. If given future completes
+ * exceptionally, the exception will be propagated to {@link Subscriber#onError(Throwable)}.
+ *
+ * <p>This particular adapter will drain iterator on the same thread that requested the
+ * entries, so it's better to avoid using long or blocking operations inside provided iterable.
+ *
+ * @param iterableFuture An iterable enclosed to a future object.
+ * @param <T> Type of the entries this publisher will emit.
+ * @return Publisher created from the given iterable.
+ */
+ public static <T> Publisher<T> fromIterable(CompletableFuture<Iterable<T>> iterableFuture) {
+ return new IterableToPublisherAdapter<>(iterableFuture, Runnable::run, Integer.MAX_VALUE);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/subscription/IterableToPublisherAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/util/subscription/IterableToPublisherAdapter.java
index 60670d3924..7318b556a4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/subscription/IterableToPublisherAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/subscription/IterableToPublisherAdapter.java
@@ -21,6 +21,7 @@ import java.lang.invoke.MethodHandles;
import java.lang.invoke.MethodHandles.Lookup;
import java.lang.invoke.VarHandle;
import java.util.Iterator;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
@@ -34,30 +35,30 @@ import java.util.concurrent.Flow.Subscription;
* @param <T> The type of the entry this publisher will emit.
*/
public class IterableToPublisherAdapter<T> implements Publisher<T> {
- private final Iterable<T> iterable;
+ private final CompletableFuture<Iterable<T>> iterableFuture;
private final Executor executor;
private final int batchSize;
/**
* Constructor.
*
- * @param iterable An iterable to issue iterator for every incoming subscription.
+ * @param iterableFuture An iterable to issue iterator for every incoming subscription.
* @param executor This executor will be used to drain iterator and supply entries to the subscription.
* @param batchSize An amount of entries to supply during a single iteration. It's always good idea
* to provide some reasonable value here in order to give am ability to other publishers which share the same
* executor to make progress.
*/
- public IterableToPublisherAdapter(Iterable<T> iterable, Executor executor, int batchSize) {
- this.iterable = iterable;
+ public IterableToPublisherAdapter(CompletableFuture<Iterable<T>> iterableFuture, Executor executor, int batchSize) {
+ this.iterableFuture = iterableFuture;
this.executor = executor;
this.batchSize = batchSize;
}
@Override
public void subscribe(Subscriber<? super T> subscriber) {
- Iterator<T> it = iterable.iterator();
+ CompletableFuture<Iterator<T>> itFuture = iterableFuture.thenApply(Iterable::iterator);
- Subscription subscription = new SubscriptionImpl<>(it, subscriber, executor, batchSize);
+ Subscription subscription = new SubscriptionImpl<>(itFuture, subscriber, executor, batchSize);
subscriber.onSubscribe(subscription);
}
@@ -80,7 +81,7 @@ public class IterableToPublisherAdapter<T> implements Publisher<T> {
}
}
- private final Iterator<T> it;
+ private final CompletableFuture<Iterator<T>> itFuture;
private final Subscriber<? super T> subscriber;
private final Executor executor;
private final int batchSize;
@@ -89,8 +90,8 @@ public class IterableToPublisherAdapter<T> implements Publisher<T> {
private long requested = 0;
private boolean wip = false;
- SubscriptionImpl(Iterator<T> it, Subscriber<? super T> subscriber, Executor executor, int batchSize) {
- this.it = it;
+ SubscriptionImpl(CompletableFuture<Iterator<T>> itFuture, Subscriber<? super T> subscriber, Executor executor, int batchSize) {
+ this.itFuture = itFuture;
this.subscriber = subscriber;
this.executor = executor;
this.batchSize = batchSize;
@@ -122,7 +123,15 @@ public class IterableToPublisherAdapter<T> implements Publisher<T> {
// the task may be scheduled several times, but it's ok, since we will deal with this
// inside task itself
- executor.execute(this::drain);
+ itFuture.whenComplete((ignored, ex) -> {
+ if (ex != null) {
+ notifyError(ex);
+
+ return;
+ }
+
+ executor.execute(this::drain);
+ });
}
@Override
@@ -141,6 +150,10 @@ public class IterableToPublisherAdapter<T> implements Publisher<T> {
long amount = amountToDrain();
+ assert itFuture.isDone();
+
+ Iterator<T> it = itFuture.join();
+
try {
while (amount-- > 0 && it.hasNext()) {
subscriber.onNext(it.next());
diff --git a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java
index 7630448c58..719813dead 100644
--- a/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java
+++ b/modules/sql-engine/src/integrationTest/java/org/apache/ignite/internal/sql/engine/ItSqlQueriesSystemViewTest.java
@@ -18,7 +18,7 @@
package org.apache.ignite.internal.sql.engine;
import static org.apache.ignite.internal.TestWrappers.unwrapIgniteImpl;
-import static org.apache.ignite.internal.sql.engine.registry.RunningQueriesRegistryImpl.SCRIPT_QUERY_TYPE;
+import static org.apache.ignite.internal.sql.engine.SqlQueriesViewProvider.SCRIPT_QUERY_TYPE;
import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.assertThrowsSqlException;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.await;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
@@ -40,7 +40,6 @@ import java.util.stream.Collectors;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.sql.SqlCommon;
-import org.apache.ignite.internal.sql.engine.registry.QueryExecutionPhase;
import org.apache.ignite.internal.sql.engine.util.MetadataMatcher;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.lang.ErrorGroups.Sql;
@@ -185,7 +184,7 @@ public class ItSqlQueriesSystemViewTest extends BaseSqlMultiStatementTest {
List<Object> row = res.get(i);
verifyQueryInfo(row, initiator.name(), SqlCommon.DEFAULT_SCHEMA_NAME, expectedQueries.get(i), timeBefore, timeAfter,
- hasLength(36), (i == 1 ? SqlQueryType.DML : SqlQueryType.QUERY).name(), i + 1);
+ hasLength(36), (i == 1 ? SqlQueryType.DML : SqlQueryType.QUERY).name(), i);
transactionIds.add((String) row.get(7));
}
@@ -239,10 +238,10 @@ public class ItSqlQueriesSystemViewTest extends BaseSqlMultiStatementTest {
timeBefore, timeAfter, hasLength(36), SqlQueryType.DML.name(), 1);
verifyQueryInfo(res.get(1), initiator.name(), SqlCommon.DEFAULT_SCHEMA_NAME, "SELECT *\nFROM `TEST`",
- timeBefore, timeAfter, equalTo((String) row.get(7)), SqlQueryType.QUERY.name(), 2);
+ timeBefore, timeAfter, equalTo((String) row.get(7)), SqlQueryType.QUERY.name(), 3);
verifyQueryInfo(res.get(2), initiator.name(), SqlCommon.DEFAULT_SCHEMA_NAME, "INSERT INTO `TEST`\nVALUES ROW(2),\nROW(3)",
- timeBefore, timeAfter, equalTo((String) row.get(7)), SqlQueryType.DML.name(), 3);
+ timeBefore, timeAfter, equalTo((String) row.get(7)), SqlQueryType.DML.name(), 4);
for (AsyncSqlCursor<InternalSqlRow> cursor : cursors) {
await(cursor.closeAsync());
@@ -253,7 +252,6 @@ public class ItSqlQueriesSystemViewTest extends BaseSqlMultiStatementTest {
}
@Test
- @SuppressWarnings("ThrowableNotThrown")
public void checkCleanupOnError() throws InterruptedException {
// Parsing error.
{
@@ -330,7 +328,7 @@ public class ItSqlQueriesSystemViewTest extends BaseSqlMultiStatementTest {
assertThat((String) row.get(idx++), hasLength(36));
// PHASE
- assertThat(row.get(idx++), equalTo(QueryExecutionPhase.EXECUTION.name()));
+ assertThat(row.get(idx++), equalTo("EXECUTION"));
// TYPE
assertThat(row.get(idx++), equalTo(queryType));
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/InflightTransactionTracker.java
similarity index 52%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/InflightTransactionTracker.java
index ebc857bf9f..f3e3d492b1 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/InflightTransactionTracker.java
@@ -15,24 +15,29 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine;
+
+import java.util.UUID;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
+import org.apache.ignite.internal.tx.impl.TransactionInflights;
/**
- * Phases of query execution.
+ * Simple facade that propagates invocations to underlying {@link TransactionInflights} object.
*/
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
+class InflightTransactionTracker implements TransactionTracker {
+ private final TransactionInflights delegate;
+
+ InflightTransactionTracker(TransactionInflights delegate) {
+ this.delegate = delegate;
+ }
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
+ @Override
+ public boolean register(UUID txId, boolean readOnly) {
+ return delegate.addInflight(txId, readOnly);
+ }
- /**
- * Query execution.
- */
- EXECUTION
+ @Override
+ public void unregister(UUID txId) {
+ delegate.removeInflight(txId);
+ }
}
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 ce5bc3ee76..65ce14b2b0 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
@@ -18,26 +18,16 @@
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.
*/
public class QueryCancel {
- private final List<Cancellable> cancelActions = new ArrayList<>(3);
-
- private Reason reason;
-
- private volatile CompletableFuture<Void> timeoutFut;
+ private final CompletableFuture<Reason> state = new CompletableFuture<>();
/**
* Adds a cancel action. If operation has already been canceled, throws a {@link QueryCancelledException}.
@@ -46,37 +36,37 @@ public class QueryCancel {
* and then throw a {@link QueryCancelledException}.
*
* @param clo Cancel action.
+ * @throws QueryCancelledException If operation has been already cancelled.
*/
- public synchronized void add(Cancellable clo) throws QueryCancelledException {
+ public void add(Cancellable clo) throws QueryCancelledException {
assert clo != null;
- if (reason != null) {
- boolean timeout = reason == Reason.TIMEOUT;
+ state.thenAccept(reason -> clo.cancel(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
- }
+ if (state.isDone()) {
+ Reason reason = state.join();
- String message = timeout ? QueryCancelledException.TIMEOUT_MSG : QueryCancelledException.CANCEL_MSG;
- throw new QueryCancelledException(message);
+ throwException(reason);
}
-
- cancelActions.add(clo);
}
/**
- * Removes the given callback.
+ * Attach child cancellation to the current one. If this has already been canceled, throws a {@link QueryCancelledException}.
*
- * @param clo Callback.
+ * <p>NOTE: If the operation is cancelled, this method will immediately trigger cancellation of a given the given cancellation
+ * and then throw a {@link QueryCancelledException}.
+ *
+ * @param another Another cancellation.
+ * @throws QueryCancelledException If operation has been already cancelled.
*/
- public synchronized void remove(Cancellable clo) {
- assert clo != null;
+ public void attach(QueryCancel another) throws QueryCancelledException {
+ state.thenAccept(another.state::complete);
- cancelActions.remove(clo);
+ if (state.isDone()) {
+ Reason reason = state.join();
+
+ throwException(reason);
+ }
}
/**
@@ -85,80 +75,29 @@ public class QueryCancel {
*
* @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;
+ public void setTimeout(ScheduledExecutorService scheduler, long timeoutMillis) {
+ scheduler.schedule(() -> state.complete(Reason.TIMEOUT), timeoutMillis, MILLISECONDS);
}
/**
* Executes cancel closure.
*/
- public synchronized void cancel() {
- doCancel(Reason.CANCEL);
+ public void cancel() {
+ state.complete(Reason.CANCEL);
}
/** Returns {@code true} if the cancellation procedure has already been started. */
public synchronized boolean isCancelled() {
- return reason != null;
+ return state.isDone();
}
- private void doCancel(Reason reason) {
- if (this.reason != null) {
- return;
- }
-
- boolean timeout = reason == Reason.TIMEOUT;
- this.reason = reason;
-
- IgniteInternalException ex = null;
-
- // Run actions in the reverse order.
- for (int i = cancelActions.size() - 1; i >= 0; i--) {
- try {
- Cancellable act = cancelActions.get(i);
-
- act.cancel(timeout);
- } catch (Exception e) {
- if (ex == null) {
- ex = new IgniteInternalException(INTERNAL_ERR, e);
- } else {
- ex.addSuppressed(e);
- }
- }
- }
-
- if (ex != null) {
- throw ex;
- }
+ private static void throwException(Reason reason) {
+ throw new QueryCancelledException(
+ reason == Reason.TIMEOUT
+ ? QueryCancelledException.TIMEOUT_MSG
+ : QueryCancelledException.CANCEL_MSG
+ );
}
enum Reason {
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 fea2eea3f1..8067878b07 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
@@ -21,10 +21,11 @@ import static java.util.Objects.requireNonNull;
import java.time.ZoneId;
import java.util.UUID;
+import java.util.function.Consumer;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueryInfo;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.util.ArrayUtils;
import org.jetbrains.annotations.Nullable;
@@ -44,7 +45,7 @@ public final class SqlOperationContext {
private final @Nullable QueryCancel cancel;
private final @Nullable String defaultSchemaName;
private final @Nullable PrefetchCallback prefetchCallback;
- private final @Nullable RunningQueryInfo queryInfo;
+ private final @Nullable Consumer<InternalTransaction> txUsedListener;
/**
* Private constructor, used by a builder.
@@ -58,7 +59,7 @@ public final class SqlOperationContext {
@Nullable QueryCancel cancel,
@Nullable String defaultSchemaName,
@Nullable PrefetchCallback prefetchCallback,
- @Nullable RunningQueryInfo queryInfo
+ @Nullable Consumer<InternalTransaction> txUsedListener
) {
this.queryId = queryId;
this.timeZoneId = timeZoneId;
@@ -68,7 +69,7 @@ public final class SqlOperationContext {
this.cancel = cancel;
this.defaultSchemaName = defaultSchemaName;
this.prefetchCallback = prefetchCallback;
- this.queryInfo = queryInfo;
+ this.txUsedListener = txUsedListener;
}
public static Builder builder() {
@@ -127,12 +128,12 @@ public final class SqlOperationContext {
}
/**
- * Returns info about running query.
- *
- * <p>May be null on remote side, but never null on node initiator.
+ * Notifies context that transaction was used for query execution.
*/
- public @Nullable RunningQueryInfo queryInfo() {
- return queryInfo;
+ public void notifyTxUsed(InternalTransaction tx) {
+ if (txUsedListener != null) {
+ txUsedListener.accept(tx);
+ }
}
/**
@@ -155,8 +156,8 @@ public final class SqlOperationContext {
private Object[] parameters = ArrayUtils.OBJECT_EMPTY_ARRAY;
private HybridTimestamp operationTime;
private @Nullable QueryTransactionContext txContext;
- private @Nullable RunningQueryInfo queryInfo;
+ private @Nullable Consumer<InternalTransaction> txUsedListener;
private @Nullable QueryCancel cancel;
private @Nullable String defaultSchemaName;
private @Nullable PrefetchCallback prefetchCallback;
@@ -201,8 +202,8 @@ public final class SqlOperationContext {
return this;
}
- public Builder queryInfo(RunningQueryInfo queryInfo) {
- this.queryInfo = queryInfo;
+ public Builder txUsedListener(Consumer<InternalTransaction> txUsedListener) {
+ this.txUsedListener = txUsedListener;
return this;
}
@@ -217,7 +218,7 @@ public final class SqlOperationContext {
cancel,
defaultSchemaName,
prefetchCallback,
- queryInfo
+ txUsedListener
);
}
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueriesViewProvider.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueriesViewProvider.java
new file mode 100644
index 0000000000..e2c838581e
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/SqlQueriesViewProvider.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine;
+
+import static org.apache.ignite.internal.type.NativeTypes.stringOf;
+
+import java.time.Instant;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Flow.Publisher;
+import org.apache.ignite.internal.sql.engine.exec.fsm.ExecutionPhase;
+import org.apache.ignite.internal.sql.engine.exec.fsm.QueryExecutor;
+import org.apache.ignite.internal.sql.engine.exec.fsm.QueryInfo;
+import org.apache.ignite.internal.systemview.api.SystemView;
+import org.apache.ignite.internal.systemview.api.SystemViews;
+import org.apache.ignite.internal.type.NativeType;
+import org.apache.ignite.internal.type.NativeTypes;
+import org.apache.ignite.internal.util.SubscriptionUtils;
+import org.jetbrains.annotations.Nullable;
+
+/** Provider that creates system view exposing queries running on a node. */
+public class SqlQueriesViewProvider {
+ public static final String SCRIPT_QUERY_TYPE = "SCRIPT";
+
+ private final CompletableFuture<QueryExecutor> queryExecutorFuture = new CompletableFuture<>();
+
+ /** Initializes provided with query executor used as datasource of running queries. */
+ public void init(QueryExecutor queryExecutor) {
+ queryExecutorFuture.complete(queryExecutor);
+ }
+
+ /** Returns system view exposing running queries. */
+ public SystemView<?> get() {
+ Publisher<QueryInfo> viewDataPublisher = SubscriptionUtils.fromIterable(
+ queryExecutorFuture.thenApply(queryExecutor -> () -> queryExecutor.runningQueries().iterator())
+ );
+
+ NativeType stringType = stringOf(Short.MAX_VALUE);
+ NativeType idType = stringOf(36);
+
+ NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
+
+ return SystemViews.<QueryInfo>nodeViewBuilder()
+ .name("SQL_QUERIES")
+ .nodeNameColumnAlias("INITIATOR_NODE")
+ .<String>addColumn("ID", idType, info -> mapId(info.id()))
+ .<String>addColumn("PHASE", stringOf(10), info -> mapPhase(info.phase()))
+ .<String>addColumn("TYPE", stringOf(10), SqlQueriesViewProvider::deriveQueryType)
+ .<String>addColumn("SCHEMA", stringType, QueryInfo::schema)
+ .<String>addColumn("SQL", stringType, QueryInfo::sql)
+ .<Instant>addColumn("START_TIME", timestampType, QueryInfo::startTime)
+ .<String>addColumn("TRANSACTION_ID", idType, info -> mapId(info.transactionId()))
+ .<String>addColumn("PARENT_ID", idType, info -> mapId(info.parentId()))
+ .<Integer>addColumn("STATEMENT_NUM", NativeTypes.INT32, info -> mapStatementNum(info.statementNum()))
+ .dataProvider(viewDataPublisher)
+ .build();
+ }
+
+ private static @Nullable String mapId(@Nullable UUID id) {
+ return id == null ? null : id.toString();
+ }
+
+ private static @Nullable String deriveQueryType(QueryInfo info) {
+ if (info.script()) {
+ return SCRIPT_QUERY_TYPE;
+ }
+
+ SqlQueryType queryType = info.queryType();
+
+ return queryType == null ? null : queryType.toString();
+ }
+
+ private static String mapPhase(ExecutionPhase phase) {
+ switch (phase) {
+ case REGISTERED: // fall through
+ case PARSING:
+ return "INITIALIZATION";
+ case OPTIMIZING:
+ return "OPTIMIZATION";
+ case CURSOR_INITIALIZATION: // fall through
+ case SCRIPT_INITIALIZATION: // fall through
+ case EXECUTING:
+ return "EXECUTION";
+ case TERMINATED:
+ return "TERMINATED";
+ default:
+ throw new AssertionError("Unknown phase " + phase);
+ }
+ }
+
+ private static @Nullable Integer mapStatementNum(int statementNum) {
+ return statementNum >= 0 ? statementNum : null;
+ }
+}
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 6259a93df5..cfca789ce0 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,35 +19,23 @@ 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.sql.engine.exec.fsm.ValidationHelper.validateDynamicParameters;
+import static org.apache.ignite.internal.sql.engine.exec.fsm.ValidationHelper.validateParsedStatement;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
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;
import java.util.Collections;
import java.util.List;
-import java.util.Objects;
-import java.util.Queue;
-import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentLinkedQueue;
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;
import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.cluster.management.topology.api.LogicalTopologyService;
import org.apache.ignite.internal.failure.FailureManager;
@@ -64,28 +52,25 @@ import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.replicator.ReplicaService;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaSyncService;
-import org.apache.ignite.internal.sql.ResultSetMetadataImpl;
import org.apache.ignite.internal.sql.SqlCommon;
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.exec.AsyncDataCursor;
import org.apache.ignite.internal.sql.engine.exec.ExchangeServiceImpl;
import org.apache.ignite.internal.sql.engine.exec.ExecutableTableRegistryImpl;
import org.apache.ignite.internal.sql.engine.exec.ExecutionDependencyResolverImpl;
-import org.apache.ignite.internal.sql.engine.exec.ExecutionService;
import org.apache.ignite.internal.sql.engine.exec.ExecutionServiceImpl;
import org.apache.ignite.internal.sql.engine.exec.LifecycleAware;
import org.apache.ignite.internal.sql.engine.exec.MailboxRegistryImpl;
import org.apache.ignite.internal.sql.engine.exec.QueryTaskExecutor;
import org.apache.ignite.internal.sql.engine.exec.QueryTaskExecutorImpl;
import org.apache.ignite.internal.sql.engine.exec.SqlRowHandler;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
import org.apache.ignite.internal.sql.engine.exec.exp.func.TableFunctionRegistryImpl;
+import org.apache.ignite.internal.sql.engine.exec.fsm.ExecutionPhase;
+import org.apache.ignite.internal.sql.engine.exec.fsm.QueryExecutor;
import org.apache.ignite.internal.sql.engine.exec.mapping.MappingServiceImpl;
import org.apache.ignite.internal.sql.engine.message.MessageServiceImpl;
-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.MultiStepPlan;
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.QueryMetadata;
@@ -93,26 +78,14 @@ 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.property.SqlProperties;
import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
-import org.apache.ignite.internal.sql.engine.registry.QueryExecutionPhase;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueriesRegistry;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueriesRegistryImpl;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueryInfo;
-import org.apache.ignite.internal.sql.engine.registry.RunningScriptInfoTracker;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManager;
import org.apache.ignite.internal.sql.engine.schema.SqlSchemaManagerImpl;
import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
-import org.apache.ignite.internal.sql.engine.sql.ParserService;
import org.apache.ignite.internal.sql.engine.sql.ParserServiceImpl;
import org.apache.ignite.internal.sql.engine.statistic.SqlStatisticManager;
import org.apache.ignite.internal.sql.engine.statistic.SqlStatisticManagerImpl;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContextImpl;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
-import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.apache.ignite.internal.sql.engine.util.Commons;
-import org.apache.ignite.internal.sql.engine.util.IteratorToDataCursorAdapter;
-import org.apache.ignite.internal.sql.engine.util.TypeUtils;
-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.metrics.SqlClientMetricSource;
@@ -127,7 +100,6 @@ import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.internal.util.ExceptionUtils;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.sql.ResultSetMetadata;
import org.apache.ignite.sql.SqlException;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -159,11 +131,7 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
private static final long EXECUTION_SERVICE_SHUTDOWN_TIMEOUT = 60_000;
- private final ParserService parserService = new ParserServiceImpl();
- private final Cache<String, ParsedResult> queryToParsedResultCache = CACHE_FACTORY.create(PARSED_RESULT_CACHE_SIZE);
-
- private static final ResultSetMetadata EMPTY_RESULT_SET_METADATA =
- new ResultSetMetadataImpl(Collections.emptyList());
+ private final SqlQueriesViewProvider queriesViewProvider = new SqlQueriesViewProvider();
private final List<LifecycleAware> services = new ArrayList<>();
@@ -191,9 +159,9 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
private final SystemViewManager systemViewManager;
- private volatile QueryTaskExecutor taskExecutor;
+ private volatile QueryExecutor queryExecutor;
- private volatile ExecutionService executionSrvc;
+ private volatile QueryTaskExecutor taskExecutor;
private volatile PrepareService prepareSvc;
@@ -221,12 +189,10 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
private final TxManager txManager;
- private final TransactionInflights transactionInflights;
+ private final TransactionTracker txTracker;
private final ScheduledExecutorService commonScheduler;
- private final RunningQueriesRegistry runningQueriesRegistry = new RunningQueriesRegistryImpl();
-
/** Constructor. */
public SqlQueryProcessor(
ClusterService clusterSrvc,
@@ -266,7 +232,7 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
this.placementDriver = placementDriver;
this.clusterCfg = clusterCfg;
this.nodeCfg = nodeCfg;
- this.transactionInflights = transactionInflights;
+ this.txTracker = new InflightTransactionTracker(transactionInflights);
this.txManager = txManager;
this.commonScheduler = commonScheduler;
sqlStatisticManager = new SqlStatisticManagerImpl(tableManager, catalogManager, lowWaterMark);
@@ -286,7 +252,7 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
taskExecutor = registerService(new QueryTaskExecutorImpl(nodeName, nodeCfg.execution().threadCount().value(), failureManager));
var mailboxRegistry = registerService(new MailboxRegistryImpl());
- SqlClientMetricSource sqlClientMetricSource = new SqlClientMetricSource(runningQueriesRegistry::openedCursorsCount);
+ SqlClientMetricSource sqlClientMetricSource = new SqlClientMetricSource(this::openedCursors);
metricManager.registerSource(sqlClientMetricSource);
metricManager.enable(sqlClientMetricSource);
@@ -365,13 +331,28 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
EXECUTION_SERVICE_SHUTDOWN_TIMEOUT
));
+ queryExecutor = registerService(new QueryExecutor(
+ CACHE_FACTORY,
+ PARSED_RESULT_CACHE_SIZE,
+ new ParserServiceImpl(),
+ taskExecutor,
+ commonScheduler,
+ clockService,
+ schemaSyncService,
+ prepareSvc,
+ catalogManager,
+ executionSrvc,
+ DEFAULT_PROPERTIES,
+ txTracker
+ ));
+
+ queriesViewProvider.init(queryExecutor);
+
clusterSrvc.topologyService().addEventHandler(executionSrvc);
clusterSrvc.topologyService().addEventHandler(mailboxRegistry);
registerService(sqlStatisticManager);
- this.executionSrvc = executionSrvc;
-
services.forEach(LifecycleAware::start);
return nullCompletedFuture();
@@ -386,11 +367,6 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
busyLock.block();
- runningQueriesRegistry.queries().stream()
- .map(RunningQueryInfo::cursor)
- .filter(Objects::nonNull)
- .forEach(AsyncSqlCursor::closeAsync);
-
metricManager.unregisterSource(SqlClientMetricSource.NAME);
List<LifecycleAware> services = new ArrayList<>(this.services);
@@ -439,25 +415,15 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
}
try {
- SqlProperties properties0 = SqlPropertiesHelper.chain(properties, DEFAULT_PROPERTIES);
QueryTransactionContext txContext = new QueryTransactionContextImpl(txManager, observableTimeTracker, transaction,
- transactionInflights);
+ txTracker);
- if (Commons.isMultiStatementQueryAllowed(properties0)) {
- return queryScript(properties0, txContext, qry, params);
- } else {
- return querySingle(properties0, txContext, qry, params);
- }
+ return queryExecutor.executeQuery(properties, txContext, qry, params);
} finally {
busyLock.leaveBusy();
}
}
- @Override
- public List<SystemView<?>> systemViews() {
- return List.of(runningQueriesRegistry.asSystemView());
- }
-
private <T extends LifecycleAware> T registerService(T service) {
services.add(service);
@@ -479,7 +445,7 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
queryCancel.setTimeout(commonScheduler, queryTimeout);
}
- ParsedResult parsedResult = queryToParsedResultCache.get(sql);
+ ParsedResult parsedResult = queryExecutor.lookupParsedResultInCache(sql);
CompletableFuture<ParsedResult> start = parsedResult != null
? completedFuture(parsedResult)
@@ -495,113 +461,20 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
queryCancel, params)
.thenApply(plan -> new QueryMetadata(plan.metadata(), plan.parameterMetadata()));
- CompletableFuture<Void> timeoutFut = queryCancel.timeoutFuture();
+ try {
+ queryCancel.add(timeout -> {
+ String message = timeout ? QueryCancelledException.TIMEOUT_MSG : QueryCancelledException.CANCEL_MSG;
- if (timeoutFut != null) {
- timeoutFut.thenAccept((r) -> {
- SqlException timeoutErr = new SqlException(EXECUTION_CANCELLED_ERR, QueryCancelledException.TIMEOUT_MSG);
- f.completeExceptionally(timeoutErr);
+ f.completeExceptionally(new SqlException(EXECUTION_CANCELLED_ERR, message));
});
+ } catch (QueryCancelledException ignored) {
+ // no-op
}
return f;
});
}
- private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> querySingle(
- SqlProperties properties,
- QueryTransactionContext txContext,
- String sql,
- Object... params
- ) {
- String schemaName = properties.get(QueryProperty.DEFAULT_SCHEMA);
- ZoneId timeZoneId = properties.get(QueryProperty.TIME_ZONE_ID);
- Long queryTimeout = properties.get(QueryProperty.QUERY_TIMEOUT);
-
- RunningQueryInfo queryInfo = runningQueriesRegistry.registerQuery(schemaName, sql, txContext.explicitTx());
-
- QueryCancel queryCancel = new QueryCancel();
-
- if (queryTimeout != 0) {
- queryCancel.setTimeout(commonScheduler, queryTimeout);
- }
-
- ParsedResult parsedResult = queryToParsedResultCache.get(sql);
-
- CompletableFuture<ParsedResult> start = parsedResult != null
- ? completedFuture(parsedResult)
- : CompletableFuture.supplyAsync(() -> parseAndCache(sql), taskExecutor);
-
- return start.thenCompose(result -> {
- queryInfo.queryType(result.queryType());
- validateParsedStatement(properties, result);
- validateDynamicParameters(result.dynamicParamsCount(), params, true);
-
- HybridTimestamp operationTime = deriveOperationTime(txContext);
-
- SqlOperationContext operationContext = SqlOperationContext.builder()
- .queryId(queryInfo.queryId())
- .cancel(queryCancel)
- .prefetchCallback(new PrefetchCallback())
- .parameters(params)
- .timeZoneId(timeZoneId)
- .defaultSchemaName(schemaName)
- .operationTime(operationTime)
- .txContext(txContext)
- .queryInfo(queryInfo)
- .build();
-
- return executeParsedStatement(operationContext, result, null);
- })
- .whenComplete((r, e) -> {
- if (e != null) {
- runningQueriesRegistry.unregister(queryInfo.queryId());
- }
- });
- }
-
- private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> queryScript(
- SqlProperties properties,
- QueryTransactionContext txCtx,
- String sql,
- Object... params
- ) {
- String schemaName = properties.get(QueryProperty.DEFAULT_SCHEMA);
- ZoneId timeZoneId = properties.get(QueryProperty.TIME_ZONE_ID);
- Long queryTimeout = properties.get(QueryProperty.QUERY_TIMEOUT);
-
- ScriptTransactionContext scriptTxContext = new ScriptTransactionContext(txCtx, transactionInflights);
- RunningScriptInfoTracker scriptInfoTracker = runningQueriesRegistry.registerScript(schemaName, sql, scriptTxContext);
-
- Instant deadline;
- if (queryTimeout != 0) {
- deadline = Instant.now().plusMillis(queryTimeout);
- } else {
- deadline = null;
- }
-
- CompletableFuture<?> start = new CompletableFuture<>();
-
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> parseFut = start
- .thenApply(ignored -> parserService.parseScript(sql))
- .thenCompose(parsedResults -> {
- MultiStatementHandler handler = new MultiStatementHandler(
- schemaName, scriptTxContext, parsedResults, params, timeZoneId, scriptInfoTracker, deadline
- );
-
- return handler.processNext();
- })
- .whenComplete((r, e) -> {
- if (e != null) {
- runningQueriesRegistry.unregister(scriptInfoTracker.queryId());
- }
- });
-
- start.completeAsync(() -> null, taskExecutor);
-
- return parseFut;
- }
-
private CompletableFuture<QueryPlan> prepareParsedStatement(String schemaName,
ParsedResult parsedResult,
HybridTimestamp timestamp,
@@ -609,7 +482,7 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
Object[] params) {
return waitForMetadata(timestamp)
- .thenCompose(none -> {
+ .thenCompose(schema -> {
SqlOperationContext ctx = SqlOperationContext.builder()
.queryId(UUID.randomUUID())
// time zone is used in execution phase,
@@ -625,123 +498,10 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
});
}
- private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executeParsedStatement(
- SqlOperationContext operationContext,
- ParsedResult parsedResult,
- @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatement
- ) {
- QueryTransactionContext txContext = operationContext.txContext();
-
- assert txContext != null;
-
- ensureStatementMatchesTx(parsedResult.queryType(), txContext);
-
- HybridTimestamp operationTime = operationContext.operationTime();
-
- RunningQueryInfo queryInfo = operationContext.queryInfo();
-
- assert queryInfo != null;
-
- return waitForMetadata(operationTime)
- .thenCompose(none -> {
- queryInfo.phase(QueryExecutionPhase.OPTIMIZATION);
-
- return prepareSvc.prepareAsync(parsedResult, operationContext);
- })
- .thenCompose(plan -> {
- queryInfo.phase(QueryExecutionPhase.EXECUTION);
-
- if (txContext.explicitTx() == null) {
- // in case of implicit tx we have to update observable time to prevent tx manager to start
- // implicit transaction too much in the past where version of catalog we used to prepare the
- // plan was not yet available
- txContext.updateObservableTime(deriveMinimalRequiredTime(plan));
- }
-
- return executePlan(operationContext, plan, nextStatement);
- });
- }
-
- private HybridTimestamp deriveOperationTime(QueryTransactionContext txContext) {
- QueryTransactionWrapper txWrapper = txContext.explicitTx();
-
- if (txWrapper == null) {
- return clockService.now();
- }
-
- return txWrapper.unwrap().startTimestamp();
- }
-
private CompletableFuture<Void> waitForMetadata(HybridTimestamp timestamp) {
return schemaSyncService.waitForMetadataCompleteness(timestamp);
}
- private CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executePlan(
- SqlOperationContext ctx,
- QueryPlan plan,
- @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatement) {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
-
- try {
- AsyncDataCursor<InternalSqlRow> dataCursor = executionSrvc.executePlan(plan, ctx);
-
- SqlQueryType queryType = plan.type();
-
- PrefetchCallback prefetchCallback = ctx.prefetchCallback();
-
- assert prefetchCallback != null;
-
- AsyncSqlCursorImpl<InternalSqlRow> cursor = new AsyncSqlCursorImpl<>(
- queryType,
- plan.metadata(),
- dataCursor,
- nextStatement
- );
-
- RunningQueryInfo queryInfo = ctx.queryInfo();
-
- assert queryInfo != null;
-
- runningQueriesRegistry.registerCursor(queryInfo, cursor);
-
- QueryTransactionContext txContext = ctx.txContext();
-
- assert txContext != null;
-
- if (queryType == SqlQueryType.QUERY) {
- if (txContext.explicitTx() == null) {
- // TODO: IGNITE-20322
- // implicit transaction started by InternalTable doesn't update observableTimeTracker. At
- // this point we don't know whether tx was started by InternalTable or ExecutionService, thus
- // let's update tracker explicitly to preserve consistency
- txContext.updateObservableTime(clockService.now());
- }
-
- // preserve lazy execution for statements that only reads
- return completedFuture(cursor);
- }
-
- // for other types let's wait for the first page to make sure premature
- // close of the cursor won't cancel an entire operation
- return cursor.onFirstPageReady()
- .thenApply(none -> {
- if (txContext.explicitTx() == null) {
- // TODO: IGNITE-20322
- // implicit transaction started by InternalTable doesn't update observableTimeTracker. At
- // this point we don't know whether tx was started by InternalTable or ExecutionService, thus
- // let's update tracker explicitly to preserve consistency
- txContext.updateObservableTime(clockService.now());
- }
-
- return cursor;
- });
- } finally {
- busyLock.leaveBusy();
- }
- }
-
@TestOnly
public MetricManager metricManager() {
return metricManager;
@@ -752,350 +512,48 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
return sqlStatisticManager;
}
- /** Performs additional validation of a parsed statement. **/
- private static void validateParsedStatement(
- SqlProperties properties,
- ParsedResult parsedResult
- ) {
- Set<SqlQueryType> allowedTypes = properties.get(QueryProperty.ALLOWED_QUERY_TYPES);
- SqlQueryType queryType = parsedResult.queryType();
-
- if (parsedResult.queryType() == SqlQueryType.TX_CONTROL) {
- String message = "Transaction control statement can not be executed as an independent statement";
-
- throw new SqlException(STMT_VALIDATION_ERR, message);
- }
-
- if (!allowedTypes.contains(queryType)) {
- String message = format("Invalid SQL statement type. Expected {} but got {}", allowedTypes, queryType);
-
- throw new SqlException(STMT_VALIDATION_ERR, message);
- }
- }
-
- private static void validateDynamicParameters(int expectedParamsCount, Object[] params, boolean exactMatch) throws SqlException {
- if (exactMatch && expectedParamsCount != params.length || params.length > expectedParamsCount) {
- String message = format(
- "Unexpected number of query parameters. Provided {} but there is only {} dynamic parameter(s).",
- params.length, expectedParamsCount
- );
-
- throw new SqlException(STMT_VALIDATION_ERR, message);
- }
-
- for (Object param : params) {
- if (!TypeUtils.supportParamInstance(param)) {
- String message = format(
- "Unsupported dynamic parameter defined. Provided '{}' is not supported.", param.getClass().getName());
-
- throw new SqlException(STMT_VALIDATION_ERR, message);
- }
- }
- }
-
- /** Checks that the statement is allowed within an external/script transaction. */
- static void ensureStatementMatchesTx(SqlQueryType queryType, QueryTransactionContext txContext) {
- QueryTransactionWrapper txWrapper = txContext.explicitTx();
-
- if (txWrapper == null) {
- return;
- }
-
- if (SqlQueryType.DDL == queryType) {
- throw new SqlException(RUNTIME_ERR, "DDL doesn't support transactions.");
- }
-
- if (SqlQueryType.DML == queryType && txWrapper.unwrap().isReadOnly()) {
- throw new SqlException(RUNTIME_ERR, "DML query cannot be started by using read only transactions.");
- }
- }
-
private static boolean shouldBeCached(SqlQueryType queryType) {
return queryType == SqlQueryType.QUERY || queryType == SqlQueryType.DML;
}
private ParsedResult parseAndCache(String sql) {
- ParsedResult result = parserService.parse(sql);
+ ParsedResult result = queryExecutor.parse(sql);
if (shouldBeCached(result.queryType())) {
- queryToParsedResultCache.put(sql, result);
+ queryExecutor.updateParsedResultCache(sql, result);
}
return result;
}
/** Returns count of opened cursors. */
- @TestOnly
public int openedCursors() {
- return runningQueriesRegistry.openedCursorsCount();
+ QueryExecutor executor = queryExecutor;
+
+ if (executor == null) {
+ return 0;
+ }
+
+ return (int) executor.runningQueries().stream()
+ .filter(info -> info.phase() == ExecutionPhase.EXECUTING)
+ .count();
}
/** Returns the number of running queries. */
@TestOnly
public int runningQueries() {
- return runningQueriesRegistry.queries().size();
- }
+ QueryExecutor executor = queryExecutor;
- private class MultiStatementHandler {
- private final ZoneId timeZoneId;
- 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.
- *
- * <p>We have no isolation within the same transaction. This implies, that any changes to the data
- * will be seen during next read. Considering lazy execution of read operations, DML started after
- * SELECT operation may, in fact, outrun actual read. To address this problem, let's collect all
- * SELECT cursor in this collection and postpone following DML until at least first page is ready
- * for every collected so far cursor.
- */
- private final Queue<CompletableFuture<Void>> inFlightSelects = new ConcurrentLinkedQueue<>();
-
- private final RunningScriptInfoTracker scriptInfoTracker;
-
- MultiStatementHandler(
- String schemaName,
- ScriptTransactionContext scriptTxContext,
- List<ParsedResult> parsedResults,
- Object[] params,
- ZoneId timeZoneId,
- RunningScriptInfoTracker scriptInfoTracker,
- @Nullable Instant deadline
- ) {
- this.timeZoneId = timeZoneId;
- this.schemaName = schemaName;
- this.statements = prepareStatementsQueue(parsedResults, params);
- this.scriptTxContext = scriptTxContext;
- this.scriptInfoTracker = scriptInfoTracker;
- this.deadline = deadline;
-
- this.scriptInfoTracker.onInitializationComplete(statements.size());
+ if (executor == null) {
+ return 0;
}
- /**
- * Returns a queue. each element of which represents parameters required to execute a single statement of the script.
- */
- private Queue<ScriptStatement> prepareStatementsQueue(List<ParsedResult> parsedResults, Object[] params) {
- assert !parsedResults.isEmpty();
-
- int paramsCount = parsedResults.stream().mapToInt(ParsedResult::dynamicParamsCount).sum();
-
- validateDynamicParameters(paramsCount, params, true);
-
- ScriptStatement[] results = new ScriptStatement[parsedResults.size()];
-
- // We fill parameters in reverse order, because each script statement
- // requires a reference to the future of the next statement.
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> prevCursorFuture = null;
- for (int i = parsedResults.size() - 1; i >= 0; i--) {
- ParsedResult result = parsedResults.get(i);
-
- Object[] params0 = Arrays.copyOfRange(params, paramsCount - result.dynamicParamsCount(), paramsCount);
- paramsCount -= result.dynamicParamsCount();
-
- results[i] = new ScriptStatement(result, params0, prevCursorFuture);
- prevCursorFuture = results[i].cursorFuture;
- }
-
- return new ArrayBlockingQueue<>(results.length, false, List.of(results));
- }
-
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> processNext() {
- ScriptStatement scriptStatement = statements.poll();
-
- assert scriptStatement != null;
-
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> cursorFuture = scriptStatement.cursorFuture;
-
- try {
- // future may be already completed by concurrent cancel of all statements due to error
- // during script execution
- if (cursorFuture.isDone()) {
- return cursorFuture;
- }
-
- ParsedResult parsedResult = scriptStatement.parsedResult;
- Object[] params = scriptStatement.dynamicParams;
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCurFut = scriptStatement.nextStatementFuture;
-
- RunningQueryInfo queryInfo =
- scriptInfoTracker.registerStatement(scriptStatement.parsedResult.originalQuery(), parsedResult.queryType());
-
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> fut;
-
- if (parsedResult.queryType() == SqlQueryType.TX_CONTROL) {
- // start of a new transaction is possible only while there is no
- // other explicit transaction; commit of a transaction will wait
- // for related cursor to be closed. In other words, we have no
- // interest in in-flight selects anymore, so lets just clean this
- // collection
- if (!inFlightSelects.isEmpty()) {
- inFlightSelects.clear();
- }
-
- // Return an empty cursor.
- fut = scriptTxContext.handleControlStatement(parsedResult.parsedTree())
- .thenApply(ignored -> new AsyncSqlCursorImpl<>(
- parsedResult.queryType(),
- EMPTY_RESULT_SET_METADATA,
- 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(queryInfo.queryId())
- .cancel(queryCancel)
- .prefetchCallback(new PrefetchCallback())
- .parameters(params)
- .timeZoneId(timeZoneId)
- .defaultSchemaName(schemaName)
- .txContext(scriptTxContext)
- .operationTime(operationTime)
- .queryInfo(queryInfo)
- .build();
-
- fut = executeParsedStatement(operationContext, parsedResult, nextCurFut);
- }
-
- boolean implicitTx = scriptTxContext.explicitTx() == null;
-
- fut.whenComplete((cursor, ex) -> {
- if (ex != null) {
- runningQueriesRegistry.unregister(queryInfo.queryId());
-
- cancelAll(ex);
-
- return;
- }
-
- if (scriptStatement.isLastStatement()) {
- // Try to rollback script managed transaction, if any.
- scriptTxContext.rollbackUncommitted();
- } else {
- CompletableFuture<Void> triggerFuture;
- ScriptStatement nextStatement = statements.peek();
-
- if (implicitTx) {
- if (cursor.queryType() != SqlQueryType.QUERY) {
- triggerFuture = cursor.onFirstPageReady();
- } else {
- triggerFuture = nullCompletedFuture();
- }
- } else {
- if (cursor.queryType() == SqlQueryType.QUERY) {
- inFlightSelects.add(CompletableFuture.anyOf(
- cursor.onClose(), cursor.onFirstPageReady()
- ).handle((r, e) -> null));
-
- if (nextStatement != null && nextStatement.parsedResult.queryType() == SqlQueryType.DML) {
- // we need to postpone DML until first page will be ready for every SELECT operation
- // prior to that DML
- triggerFuture = CompletableFuture.allOf(inFlightSelects.toArray(CompletableFuture[]::new));
-
- inFlightSelects.clear();
- } else {
- triggerFuture = nullCompletedFuture();
- }
- } else {
- CompletableFuture<Void> prefetchFuture = cursor.onFirstPageReady();
-
- // for non query statements cursor should not be resolved until the very first page is ready.
- // if prefetch was completed exceptionally, then cursor future is expected to be completed
- // exceptionally as well, resulting in the early return in the very beginning of the `whenComplete`
- assert prefetchFuture.isDone() && !prefetchFuture.isCompletedExceptionally()
- : "prefetch future is expected to be completed successfully, but was "
- + (prefetchFuture.isDone() ? "completed exceptionally" : "not completed");
-
- triggerFuture = nullCompletedFuture();
- }
- }
-
- triggerFuture.thenRunAsync(this::processNext, taskExecutor)
- .exceptionally(e -> {
- cancelAll(e);
-
- return null;
- });
- }
- }).whenCompleteAsync((cursor, ex) -> {
- if (ex != null) {
- cursorFuture.completeExceptionally(ex);
- } else {
- cursorFuture.complete(cursor);
- }
- }, taskExecutor);
- } catch (Throwable e) {
- scriptTxContext.onError(e);
-
- cursorFuture.completeExceptionally(e);
-
- cancelAll(e);
- }
-
- return cursorFuture;
- }
-
- private void cancelAll(Throwable cause) {
- for (ScriptStatement scriptStatement : statements) {
- CompletableFuture<AsyncSqlCursor<InternalSqlRow>> fut = scriptStatement.cursorFuture;
-
- if (fut.isDone()) {
- continue;
- }
-
- fut.completeExceptionally(new SqlException(
- EXECUTION_CANCELLED_ERR,
- "The script execution was canceled due to an error in the previous statement.",
- cause
- ));
- }
- }
-
- private class ScriptStatement {
- private final CompletableFuture<AsyncSqlCursor<InternalSqlRow>> cursorFuture = new CompletableFuture<>();
- private final CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatementFuture;
- private final ParsedResult parsedResult;
- private final Object[] dynamicParams;
-
- private ScriptStatement(
- ParsedResult parsedResult,
- Object[] dynamicParams,
- @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatementFuture
- ) {
- this.parsedResult = parsedResult;
- this.dynamicParams = dynamicParams;
- this.nextStatementFuture = nextStatementFuture;
- }
+ return executor.runningQueries().size();
+ }
- boolean isLastStatement() {
- return nextStatementFuture == null;
- }
- }
+ @Override
+ public List<SystemView<?>> systemViews() {
+ return List.of(queriesViewProvider.get());
}
/** Completes the provided future when the callback is called. */
@@ -1115,26 +573,4 @@ public class SqlQueryProcessor implements QueryProcessor, SystemViewProvider {
return prefetchFuture;
}
}
-
- private HybridTimestamp deriveMinimalRequiredTime(QueryPlan plan) {
- Integer catalogVersion = null;
-
- if (plan instanceof MultiStepPlan) {
- catalogVersion = ((MultiStepPlan) plan).catalogVersion();
- } else if (plan instanceof KeyValueModifyPlan) {
- catalogVersion = ((KeyValueModifyPlan) plan).catalogVersion();
- } else if (plan instanceof KeyValueGetPlan) {
- catalogVersion = ((KeyValueGetPlan) plan).catalogVersion();
- }
-
- if (catalogVersion != null) {
- Catalog catalog = catalogManager.catalog(catalogVersion);
-
- assert catalog != null;
-
- return HybridTimestamp.hybridTimestamp(catalog.time());
- }
-
- return clockService.now();
- }
}
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 af1044683c..f5c608d088 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
@@ -30,7 +30,6 @@ import java.util.Objects;
import java.util.TimeZone;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import org.apache.calcite.DataContext;
@@ -40,6 +39,7 @@ import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.RunnableX;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.exec.exp.ExpressionFactory;
import org.apache.ignite.internal.sql.engine.exec.exp.ExpressionFactoryImpl;
@@ -96,7 +96,7 @@ public class ExecutionContext<RowT> implements DataContext {
private SharedState sharedState = new SharedState();
- private final @Nullable CompletableFuture<Void> timeoutFut;
+ private final @Nullable QueryCancel cancel;
/**
* Constructor.
@@ -110,7 +110,7 @@ public class ExecutionContext<RowT> implements DataContext {
* @param params Parameters.
* @param txAttributes Transaction attributes.
* @param timeZoneId Session time-zone ID.
- * @param timeoutFut Timeout future.
+ * @param cancel Cancellation handle.
*/
@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
public ExecutionContext(
@@ -123,7 +123,7 @@ public class ExecutionContext<RowT> implements DataContext {
Map<String, Object> params,
TxAttributes txAttributes,
ZoneId timeZoneId,
- @Nullable CompletableFuture<Void> timeoutFut
+ @Nullable QueryCancel cancel
) {
this.executor = executor;
this.qryId = qryId;
@@ -134,7 +134,7 @@ public class ExecutionContext<RowT> implements DataContext {
this.originatingNodeName = originatingNodeName;
this.txAttributes = txAttributes;
this.timeZoneId = timeZoneId;
- this.timeoutFut = timeoutFut;
+ this.cancel = cancel;
expressionFactory = new ExpressionFactoryImpl<>(
this,
@@ -394,16 +394,17 @@ public class ExecutionContext<RowT> implements DataContext {
* if timeout is set of this context.
*/
public void scheduleTimeout(CompletableFuture<?> fut) {
- if (timeoutFut == null) {
+ if (cancel == null) {
return;
}
- Executor executor = task -> execute(task::run, (err) -> {});
+ cancel.add(timeout -> {
+ if (!timeout) {
+ return;
+ }
- timeoutFut.thenAcceptAsync(
- (r) -> fut.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG)),
- executor
- );
+ fut.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG));
+ });
}
/** Creates {@link PartitionProvider} for the given source table. */
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 34383970c5..229cd17c86 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
@@ -101,7 +101,6 @@ import org.apache.ignite.internal.sql.engine.prepare.Fragment;
import org.apache.ignite.internal.sql.engine.prepare.IgniteRelShuttle;
import org.apache.ignite.internal.sql.engine.prepare.MultiStepPlan;
import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueryInfo;
import org.apache.ignite.internal.sql.engine.rel.IgniteIndexScan;
import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
import org.apache.ignite.internal.sql.engine.rel.IgniteTableModify;
@@ -311,11 +310,6 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
queryManager.close(reason);
});
- CompletableFuture<Void> timeoutFut = cancelHandler.timeoutFuture();
- if (timeoutFut != null) {
- timeoutFut.thenAcceptAsync((r) -> queryManager.close(QueryCompletionReason.TIMEOUT), taskExecutor);
- }
-
QueryTransactionContext txContext = operationContext.txContext();
assert txContext != null;
@@ -325,13 +319,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
AsyncCursor<InternalSqlRow> dataCursor = queryManager.execute(tx, plan);
- if (txWrapper.implicit()) {
- RunningQueryInfo queryInfo = operationContext.queryInfo();
-
- assert queryInfo != null;
-
- queryInfo.transactionId(tx.id());
- }
+ operationContext.notifyTxUsed(tx);
PrefetchCallback prefetchCallback = operationContext.prefetchCallback();
@@ -429,7 +417,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
Commons.parametersMap(operationContext.parameters()),
TxAttributes.dummy(),
operationContext.timeZoneId(),
- queryCancel.timeoutFuture()
+ operationContext.cancel()
);
QueryTransactionContext txContext = operationContext.txContext();
@@ -474,12 +462,13 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
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);
- }
+ queryCancel.add(timeout -> {
+ if (!timeout) {
+ return;
+ }
+
+ ret.completeExceptionally(new QueryCancelledException(QueryCancelledException.TIMEOUT_MSG));
+ });
return new IteratorToDataCursorAdapter<>(ret, Runnable::run);
}
@@ -785,7 +774,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
private final @Nullable CompletableFuture<AsyncRootNode<RowT, InternalSqlRow>> root;
- private final @Nullable CompletableFuture<Void> timeoutFut;
+ private final @Nullable QueryCancel cancel;
/** Mutex for {@link #remoteFragmentInitCompletion} modifications. */
private final Object initMux = new Object();
@@ -802,9 +791,6 @@ 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 -> {
@@ -820,10 +806,10 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
});
this.root = root;
- this.timeoutFut = queryCancel.timeoutFuture();
+ this.cancel = ctx.cancel();
} else {
this.root = null;
- this.timeoutFut = null;
+ this.cancel = null;
}
}
@@ -937,7 +923,7 @@ public class ExecutionServiceImpl<RowT> implements ExecutionService, TopologyEve
Commons.parametersMap(ctx.parameters()),
txAttributes,
ctx.timeZoneId(),
- timeoutFut
+ cancel
);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TransactionTracker.java
similarity index 69%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TransactionTracker.java
index ebc857bf9f..d4082bce3f 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/TransactionTracker.java
@@ -15,24 +15,17 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine.exec;
+
+import java.util.UUID;
/**
- * Phases of query execution.
+ * Tracker of ongoing transactions.
*/
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
-
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
+public interface TransactionTracker {
+ /** Registers given transaction within the tracker. */
+ boolean register(UUID txId, boolean readOnly);
- /**
- * Query execution.
- */
- EXECUTION
+ /** De-registers given transaction from the tracker. */
+ void unregister(UUID txId);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java
new file mode 100644
index 0000000000..b860b20735
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/CursorInitializationPhaseHandler.java
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursorImpl;
+import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursor;
+import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+
+/** Handler that acquires data cursor and saves it to {@link Query query state}. */
+class CursorInitializationPhaseHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new CursorInitializationPhaseHandler();
+
+ private CursorInitializationPhaseHandler() { }
+
+ @Override
+ public Result handle(Query query) {
+ QueryPlan plan = query.plan;
+ SqlOperationContext context = query.operationContext;
+
+ assert plan != null;
+ assert context != null;
+
+ AsyncDataCursor<InternalSqlRow> dataCursor = query.executor.executePlan(context, plan);
+
+ SqlQueryType queryType = plan.type();
+
+ PrefetchCallback prefetchCallback = context.prefetchCallback();
+
+ assert prefetchCallback != null;
+
+ AsyncSqlCursorImpl<InternalSqlRow> cursor = new AsyncSqlCursorImpl<>(
+ queryType,
+ plan.metadata(),
+ dataCursor,
+ query.nextCursorFuture
+ );
+
+ query.cursor = cursor;
+
+ QueryTransactionContext txContext = query.txContext;
+
+ assert txContext != null;
+
+ if (queryType == SqlQueryType.QUERY) {
+ if (txContext.explicitTx() == null) {
+ // TODO: IGNITE-20322
+ // implicit transaction started by InternalTable doesn't update observableTimeTracker. At
+ // this point we don't know whether tx was started by InternalTable or ExecutionService, thus
+ // let's update tracker explicitly to preserve consistency
+ txContext.updateObservableTime(query.executor.clockNow());
+ }
+
+ query.moveTo(ExecutionPhase.EXECUTING);
+
+ // Postpone scheduling for TERMINATED phase to not to miss EXECUTING phase,
+ // because cursor may be already closed at this point.
+ cursor.onClose()
+ .whenComplete((ignored, ex) -> query.moveTo(ExecutionPhase.TERMINATED));
+
+ // preserve lazy execution for statements that only reads
+ return Result.proceedImmediately();
+ }
+
+ // for other types let's wait for the first page to make sure premature
+ // close of the cursor won't cancel an entire operation
+ CompletableFuture<Void> awaitFuture = cursor.onFirstPageReady()
+ .thenApply(none -> {
+ if (txContext.explicitTx() == null) {
+ // TODO: IGNITE-20322
+ // implicit transaction started by InternalTable doesn't update observableTimeTracker. At
+ // this point we don't know whether tx was started by InternalTable or ExecutionService, thus
+ // let's update tracker explicitly to preserve consistency
+ txContext.updateObservableTime(query.executor.clockNow());
+ }
+
+ query.moveTo(ExecutionPhase.EXECUTING);
+
+ // Postpone scheduling for TERMINATED phase to not to miss EXECUTING phase,
+ // because cursor may be already closed at this point.
+ cursor.onClose()
+ .whenComplete((ignored, ex) -> query.moveTo(ExecutionPhase.TERMINATED));
+
+ return null;
+ });
+
+ return Result.proceedAfter(awaitFuture);
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhase.java
new file mode 100644
index 0000000000..6720128d1b
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhase.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+/**
+ * Enumerates possible phases of query execution.
+ *
+ * <p>This enum is a sort of transition graph since generally there is exactly one transition on a normal execution path. The one exception
+ * is {@link #PARSING} phase where path diverges to {@link #OPTIMIZING} and {@link #CURSOR_INITIALIZATION}. Also, it's allowed to move to
+ * {@link #TERMINATED} phase from any other phase (usually, in case of exception).
+ *
+ * <p>Currently, there are following transitions:
+ * <pre>
+ * For single statement execution:
+ * REGISTERED --> PARSING --> OPTIMIZING --> CURSOR_INITIALIZATION --> EXECUTING --> TERMINATED
+ *
+ * For script execution:
+ * REGISTERED --> PARSING --> SCRIPT_INITIALIZATION --> EXECUTING --> TERMINATED
+ *
+ * For single statement execution spawned by script:
+ * REGISTERED --> OPTIMIZING --> CURSOR_INITIALIZATION --> EXECUTING --> TERMINATED
+ * </pre>
+ */
+public enum ExecutionPhase {
+ /** Query is registered on server. */
+ REGISTERED(RegisteredPhaseHandler.INSTANCE),
+ /** Query string is parsed at the moment. Parsed AST may or may not be available yet. */
+ PARSING(ParsingPhaseHandler.INSTANCE),
+ /** AST is available now, optimization task is submitted. */
+ OPTIMIZING(OptimizingPhaseHandler.INSTANCE),
+ /** Query has been validated, plan is ready as well. At this point plan is mapped on cluster and cursor are initialised. */
+ CURSOR_INITIALIZATION(CursorInitializationPhaseHandler.INSTANCE),
+ /** Multiple queries must be scheduled in a way to not to interfere with each other. */
+ SCRIPT_INITIALIZATION(ScriptInitializationPhaseHandler.INSTANCE),
+ /**
+ * All preparations have been done, distributed cursor is ready. Fetching data for cursor is not managed by fsm, therefore processing is
+ * temporary stopped.
+ */
+ EXECUTING(StopHandler.INSTANCE),
+ /** Query is terminated (either successfully or due to error), and being deregister soon. This is terminal phase. */
+ TERMINATED(StopHandler.INSTANCE);
+
+ private final ExecutionPhaseHandler handler;
+
+ ExecutionPhase(ExecutionPhaseHandler handler) {
+ this.handler = handler;
+ }
+
+ /** Evaluates current phase of query execution. */
+ Result evaluate(Query query) {
+ return handler.handle(query);
+ }
+
+ /** Returns {@code true} if transition is allowed from current phase to given one, return {@code false} otherwise. */
+ boolean transitionAllowed(ExecutionPhase newPhase) {
+ return ordinal() < newPhase.ordinal();
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhaseHandler.java
similarity index 70%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhaseHandler.java
index ebc857bf9f..ebc8962abf 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ExecutionPhaseHandler.java
@@ -15,24 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine.exec.fsm;
/**
- * Phases of query execution.
+ * Handler of execution phase.
+ *
+ * <p>Encapsulates actions required to be done in order to move query from one phase to another.
*/
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
-
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
-
- /**
- * Query execution.
- */
- EXECUTION
+@FunctionalInterface
+interface ExecutionPhaseHandler {
+ Result handle(Query query);
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/MultiStatementHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/MultiStatementHandler.java
new file mode 100644
index 0000000000..d046fec833
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/MultiStatementHandler.java
@@ -0,0 +1,281 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import static org.apache.ignite.internal.sql.engine.exec.fsm.ValidationHelper.validateDynamicParameters;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.lang.ErrorGroups.Sql.EXECUTION_CANCELLED_ERR;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import org.apache.ignite.internal.sql.ResultSetMetadataImpl;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursorImpl;
+import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
+import org.apache.ignite.internal.sql.engine.util.IteratorToDataCursorAdapter;
+import org.apache.ignite.sql.ResultSetMetadata;
+import org.apache.ignite.sql.SqlException;
+import org.jetbrains.annotations.Nullable;
+
+class MultiStatementHandler {
+ private static final ResultSetMetadata EMPTY_RESULT_SET_METADATA =
+ new ResultSetMetadataImpl(Collections.emptyList());
+
+ private final Query query;
+ private final Queue<ScriptStatement> statements;
+ private final ScriptTransactionContext scriptTxContext;
+
+ /**
+ * Collection is used to track SELECT statements to postpone following DML operation.
+ *
+ * <p>We have no isolation within the same transaction. This implies, that any changes to the data
+ * will be seen during next read. Considering lazy execution of read operations, DML started after SELECT operation may, in fact, outrun
+ * actual read. To address this problem, let's collect all SELECT cursor in this collection and postpone following DML until at least
+ * first page is ready for every collected so far cursor.
+ */
+ private final Queue<CompletableFuture<Void>> inFlightSelects = new ConcurrentLinkedQueue<>();
+
+ private final Queue<CompletableFuture<Void>> dependentQueries = new ConcurrentLinkedQueue<>();
+
+ MultiStatementHandler(
+ TransactionTracker txTracker,
+ Query query,
+ QueryTransactionContext txContext,
+ List<ParsedResult> parsedResults,
+ Object[] params
+ ) {
+ this.query = query;
+ this.statements = prepareStatementsQueue(parsedResults, params);
+ this.scriptTxContext = new ScriptTransactionContext(txContext, txTracker);
+ }
+
+ /**
+ * Returns a queue. each element of which represents parameters required to execute a single statement of the script.
+ */
+ private static Queue<ScriptStatement> prepareStatementsQueue(List<ParsedResult> parsedResults, Object[] params) {
+ assert !parsedResults.isEmpty();
+
+ int paramsCount = parsedResults.stream().mapToInt(ParsedResult::dynamicParamsCount).sum();
+
+ validateDynamicParameters(paramsCount, params, true);
+
+ ScriptStatement[] results = new ScriptStatement[parsedResults.size()];
+
+ // We fill parameters in reverse order, because each script statement
+ // requires a reference to the future of the next statement.
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> prevCursorFuture = null;
+ for (int i = parsedResults.size() - 1; i >= 0; i--) {
+ ParsedResult result = parsedResults.get(i);
+
+ Object[] params0 = Arrays.copyOfRange(params, paramsCount - result.dynamicParamsCount(), paramsCount);
+ paramsCount -= result.dynamicParamsCount();
+
+ results[i] = new ScriptStatement(i, result, params0, prevCursorFuture);
+ prevCursorFuture = results[i].cursorFuture;
+ }
+
+ return new ArrayBlockingQueue<>(results.length, false, List.of(results));
+ }
+
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> processNext() {
+ ScriptStatement scriptStatement = statements.poll();
+
+ assert scriptStatement != null;
+
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> cursorFuture = scriptStatement.cursorFuture;
+
+ try {
+ // future may be already completed by concurrent cancel of all statements due to error
+ // during script execution
+ if (cursorFuture.isDone()) {
+ return cursorFuture;
+ }
+
+ int statementNum = scriptStatement.idx;
+ ParsedResult parsedResult = scriptStatement.parsedResult;
+ Object[] params = scriptStatement.dynamicParams;
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCurFut = scriptStatement.nextStatementFuture;
+
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> fut;
+
+ if (parsedResult.queryType() == SqlQueryType.TX_CONTROL) {
+ // start of a new transaction is possible only while there is no
+ // other explicit transaction; commit of a transaction will wait
+ // for related cursor to be closed. In other words, we have no
+ // interest in in-flight selects anymore, so lets just clean this
+ // collection
+ if (!inFlightSelects.isEmpty()) {
+ inFlightSelects.clear();
+ }
+
+ // Return an empty cursor.
+ fut = scriptTxContext.handleControlStatement(parsedResult.parsedTree())
+ .thenApply(ignored -> new AsyncSqlCursorImpl<>(
+ parsedResult.queryType(),
+ EMPTY_RESULT_SET_METADATA,
+ new IteratorToDataCursorAdapter<>(Collections.emptyIterator()),
+ nextCurFut
+ ));
+ } else {
+ scriptTxContext.registerCursorFuture(parsedResult.queryType(), cursorFuture);
+
+ fut = query.executor.executeChildQuery(query, scriptTxContext, statementNum, parsedResult, params, nextCurFut);
+ }
+
+ boolean implicitTx = scriptTxContext.explicitTx() == null;
+
+ fut.whenComplete((cursor, ex) -> {
+ if (ex != null) {
+ scriptTxContext.onError(ex);
+
+ cursorFuture.completeExceptionally(ex);
+
+ cancelAll(ex);
+
+ return;
+ }
+
+ cursorFuture.complete(cursor);
+
+ if (!cursor.onClose().isDone()) {
+ dependentQueries.add(cursor.onClose());
+ }
+
+ if (scriptStatement.isLastStatement()) {
+ // Try to rollback script managed transaction, if any.
+ scriptTxContext.rollbackUncommitted();
+
+ // In normal execution flow (i.e. not exceptional) query should not skip phases,
+ // therefore before termination let's wait until query enter the EXECUTION phase.
+ query.onPhaseStarted(ExecutionPhase.EXECUTING)
+ .thenRun(this::scheduleTermination);
+ } else {
+ CompletableFuture<Void> triggerFuture;
+ ScriptStatement nextStatement = statements.peek();
+
+ if (implicitTx) {
+ if (cursor.queryType() != SqlQueryType.QUERY) {
+ triggerFuture = cursor.onFirstPageReady();
+ } else {
+ triggerFuture = nullCompletedFuture();
+ }
+ } else {
+ if (cursor.queryType() == SqlQueryType.QUERY) {
+ inFlightSelects.add(CompletableFuture.anyOf(
+ cursor.onClose(), cursor.onFirstPageReady()
+ ).handle((r, e) -> null));
+
+ if (nextStatement != null && nextStatement.parsedResult.queryType() == SqlQueryType.DML) {
+ // we need to postpone DML until first page will be ready for every SELECT operation
+ // prior to that DML
+ triggerFuture = CompletableFuture.allOf(inFlightSelects.toArray(CompletableFuture[]::new));
+
+ inFlightSelects.clear();
+ } else {
+ triggerFuture = nullCompletedFuture();
+ }
+ } else {
+ CompletableFuture<Void> prefetchFuture = cursor.onFirstPageReady();
+
+ // for non query statements cursor should not be resolved until the very first page is ready.
+ // if prefetch was completed exceptionally, then cursor future is expected to be completed
+ // exceptionally as well, resulting in the early return in the very beginning of the `whenComplete`
+ assert prefetchFuture.isDone() && !prefetchFuture.isCompletedExceptionally()
+ : "prefetch future is expected to be completed successfully, but was "
+ + (prefetchFuture.isDone() ? "completed exceptionally" : "not completed");
+
+ triggerFuture = nullCompletedFuture();
+ }
+ }
+
+ triggerFuture.thenRunAsync(this::processNext, query.executor::execute)
+ .exceptionally(e -> {
+ cancelAll(e);
+
+ return null;
+ });
+ }
+ });
+ } catch (Throwable e) {
+ scriptTxContext.onError(e);
+
+ cursorFuture.completeExceptionally(e);
+
+ cancelAll(e);
+ }
+
+ return cursorFuture;
+ }
+
+ private void cancelAll(Throwable cause) {
+ for (ScriptStatement scriptStatement : statements) {
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> fut = scriptStatement.cursorFuture;
+
+ if (fut.isDone()) {
+ continue;
+ }
+
+ fut.completeExceptionally(new SqlException(
+ EXECUTION_CANCELLED_ERR,
+ "The script execution was canceled due to an error in the previous statement.",
+ cause
+ ));
+ }
+
+ scheduleTermination();
+ }
+
+ private void scheduleTermination() {
+ CompletableFuture.allOf(dependentQueries.toArray(CompletableFuture[]::new))
+ .whenComplete((ignored, ex) -> query.moveTo(ExecutionPhase.TERMINATED));
+ }
+
+ private static class ScriptStatement {
+ private final CompletableFuture<AsyncSqlCursor<InternalSqlRow>> cursorFuture = new CompletableFuture<>();
+ private final CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatementFuture;
+ private final ParsedResult parsedResult;
+ private final Object[] dynamicParams;
+ private final int idx;
+
+ private ScriptStatement(
+ int idx,
+ ParsedResult parsedResult,
+ Object[] dynamicParams,
+ @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextStatementFuture
+ ) {
+ this.idx = idx;
+ this.parsedResult = parsedResult;
+ this.dynamicParams = dynamicParams;
+ this.nextStatementFuture = nextStatementFuture;
+ }
+
+ boolean isLastStatement() {
+ return nextStatementFuture == null;
+ }
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/OptimizingPhaseHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/OptimizingPhaseHandler.java
new file mode 100644
index 0000000000..d0fc0c52e9
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/OptimizingPhaseHandler.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import static org.apache.ignite.internal.sql.engine.exec.fsm.ValidationHelper.validateDynamicParameters;
+import static org.apache.ignite.internal.sql.engine.exec.fsm.ValidationHelper.validateParsedStatement;
+import static org.apache.ignite.lang.ErrorGroups.Sql.RUNTIME_ERR;
+
+import java.time.ZoneId;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.sql.engine.QueryProperty;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.SqlQueryProcessor.PrefetchCallback;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.prepare.PrepareService;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
+import org.apache.ignite.sql.SqlException;
+
+/** Validates parsed AST acquired on the previous phase and submit optimization task to {@link PrepareService}. */
+class OptimizingPhaseHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new OptimizingPhaseHandler();
+
+ private OptimizingPhaseHandler() {
+ }
+
+ @Override
+ public Result handle(Query query) {
+ ParsedResult result = query.parsedResult;
+
+ assert result != null : "Query is expected to be parsed at this phase";
+
+ validateParsedStatement(query.properties, result);
+ validateDynamicParameters(result.dynamicParamsCount(), query.params, true);
+ ensureStatementMatchesTx(result.queryType(), query.txContext);
+
+ HybridTimestamp operationTime = query.executor.deriveOperationTime(query.txContext);
+
+ String schemaName = query.properties.get(QueryProperty.DEFAULT_SCHEMA);
+ ZoneId timeZoneId = query.properties.get(QueryProperty.TIME_ZONE_ID);
+
+ SqlOperationContext operationContext = SqlOperationContext.builder()
+ .queryId(query.id)
+ .cancel(query.cancel)
+ .prefetchCallback(new PrefetchCallback())
+ .parameters(query.params)
+ .timeZoneId(timeZoneId)
+ .defaultSchemaName(schemaName)
+ .operationTime(operationTime)
+ .txContext(query.txContext)
+ .txUsedListener(tx -> query.usedTransaction = tx)
+ .build();
+
+ query.operationContext = operationContext;
+
+ CompletableFuture<Void> awaitFuture = query.executor.waitForMetadata(operationTime)
+ .thenCompose(none -> query.executor.prepare(result, operationContext)
+ .thenAccept(plan -> {
+ if (query.txContext.explicitTx() == null) {
+ // in case of implicit tx we have to update observable time to prevent tx manager to start
+ // implicit transaction too much in the past where version of catalog we used to prepare the
+ // plan was not yet available
+ query.txContext.updateObservableTime(query.executor.deriveMinimalRequiredTime(plan));
+ }
+
+ query.plan = plan;
+
+ query.moveTo(ExecutionPhase.CURSOR_INITIALIZATION);
+ }));
+
+ return Result.proceedAfter(awaitFuture);
+ }
+
+ /** Checks that the statement is allowed within an external/script transaction. */
+ private static void ensureStatementMatchesTx(SqlQueryType queryType, QueryTransactionContext txContext) {
+ QueryTransactionWrapper txWrapper = txContext.explicitTx();
+
+ if (txWrapper == null) {
+ return;
+ }
+
+ if (SqlQueryType.DDL == queryType) {
+ throw new SqlException(RUNTIME_ERR, "DDL doesn't support transactions.");
+ }
+
+ if (SqlQueryType.DML == queryType && txWrapper.unwrap().isReadOnly()) {
+ throw new SqlException(RUNTIME_ERR, "DML query cannot be started by using read only transactions.");
+ }
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ParsingPhaseHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ParsingPhaseHandler.java
new file mode 100644
index 0000000000..1bdbbc1514
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ParsingPhaseHandler.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import static org.apache.ignite.internal.sql.engine.util.Commons.isMultiStatementQueryAllowed;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+
+/** Parses the query string and populate {@link Query query state} with results. */
+class ParsingPhaseHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new ParsingPhaseHandler();
+
+ private ParsingPhaseHandler() { }
+
+ @Override
+ public Result handle(Query query) {
+ ParsedResult parsedResult = query.executor.lookupParsedResultInCache(query.sql);
+
+ if (parsedResult != null) {
+ query.parsedResult = parsedResult;
+
+ query.moveTo(ExecutionPhase.OPTIMIZING);
+
+ return Result.proceedImmediately();
+ }
+
+ CompletableFuture<Void> awaitFuture = new CompletableFuture<>();
+ query.executor.execute(() -> {
+ try {
+ ParsedResult result;
+ if (isMultiStatementQueryAllowed(query.properties)) {
+ List<ParsedResult> results = query.executor.parseScript(query.sql);
+
+ if (results.size() != 1 || results.get(0).queryType() == SqlQueryType.TX_CONTROL) {
+ // the query is script indeed, need different execution path
+ query.parsedScript = results;
+
+ query.moveTo(ExecutionPhase.SCRIPT_INITIALIZATION);
+ awaitFuture.complete(null);
+
+ return;
+ }
+
+ result = results.get(0);
+ } else {
+ result = query.executor.parse(query.sql);
+ }
+
+ if (shouldBeCached(result.queryType())) {
+ query.executor.updateParsedResultCache(query.sql, result);
+ }
+
+ query.parsedResult = result;
+
+ query.moveTo(ExecutionPhase.OPTIMIZING);
+ awaitFuture.complete(null);
+ } catch (Throwable th) {
+ awaitFuture.completeExceptionally(th);
+ }
+ });
+
+ return Result.proceedAfter(awaitFuture);
+ }
+
+ private static boolean shouldBeCached(SqlQueryType queryType) {
+ return queryType == SqlQueryType.QUERY || queryType == SqlQueryType.DML;
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Query.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Query.java
new file mode 100644
index 0000000000..6154a57958
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Query.java
@@ -0,0 +1,193 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.QueryCancel;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.exec.fsm.Result.Status;
+import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
+import org.apache.ignite.internal.sql.engine.property.SqlProperties;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Represents a query initiated on current node.
+ *
+ * <p>Encapsulates intermediate state populated throughout query lifecycle.
+ */
+class Query implements Runnable {
+ // Below are attributes the query was initialized with
+ final Instant createdAt;
+ final @Nullable UUID parentId;
+ final int statementNum;
+ final UUID id;
+ final String sql;
+ final Object[] params;
+ final QueryCancel cancel = new QueryCancel();
+ final QueryExecutor executor;
+ final SqlProperties properties;
+ final QueryTransactionContext txContext;
+ final @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCursorFuture;
+
+ // Below is volatile state populated during processing of particular stage for single statement execution
+ volatile @Nullable ParsedResult parsedResult = null;
+ volatile @Nullable SqlOperationContext operationContext = null;
+ volatile @Nullable QueryPlan plan = null;
+ volatile @Nullable InternalTransaction usedTransaction = null;
+ volatile @Nullable AsyncSqlCursor<InternalSqlRow> cursor = null;
+
+ // Below is volatile state for script processing
+ volatile @Nullable List<ParsedResult> parsedScript = null;
+
+ // Below are internal attributes
+ private final ConcurrentMap<ExecutionPhase, CompletableFuture<Void>> onPhaseStartedCallback = new ConcurrentHashMap<>();
+ private final Object mux = new Object();
+
+ private volatile ExecutionPhase currentPhase = ExecutionPhase.REGISTERED;
+
+ Query(
+ Instant createdAt,
+ QueryExecutor executor,
+ UUID id,
+ String sql,
+ SqlProperties properties,
+ QueryTransactionContext txContext,
+ Object[] params,
+ @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCursorFuture
+ ) {
+ this.createdAt = createdAt;
+ this.executor = executor;
+ this.id = id;
+ this.sql = sql;
+ this.properties = properties;
+ this.txContext = txContext;
+ this.params = params;
+ this.nextCursorFuture = nextCursorFuture;
+
+ this.parentId = null;
+ this.statementNum = -1;
+ }
+
+ Query(
+ Instant createdAt,
+ Query parent,
+ ParsedResult parsedResult,
+ int statementNum,
+ UUID id,
+ QueryTransactionContext txContext,
+ Object[] params,
+ @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCursorFuture
+ ) {
+ this.createdAt = createdAt;
+ this.executor = parent.executor;
+ this.parentId = parent.id;
+ this.statementNum = statementNum;
+ this.id = id;
+ this.sql = parsedResult.originalQuery();
+ this.properties = parent.properties;
+ this.txContext = txContext;
+ this.params = params;
+ this.nextCursorFuture = nextCursorFuture;
+
+ this.parsedResult = parsedResult;
+ }
+
+ @Override
+ public void run() {
+ Result result;
+ do {
+ ExecutionPhase phaseBefore = currentPhase;
+
+ try {
+ result = phaseBefore.evaluate(this);
+ } catch (Throwable th) {
+ // handles exception from synchronous part of phase evaluation
+
+ onError(th);
+
+ return;
+ }
+
+ if (IgniteUtils.assertionsEnabled()
+ && result.status() == Result.Status.PROCEED_IMMEDIATELY
+ && currentPhase == phaseBefore) {
+ throw new AssertionError("Attempt to immediately execute the same state. Did you forget to move query to next state?");
+ }
+
+ if (result.status() == Status.STOP) {
+ break;
+ }
+
+ if (result.status() == Status.SCHEDULE) {
+ CompletableFuture<Void> awaitFuture = result.await();
+
+ assert awaitFuture != null;
+
+ // reschedule only if required computation has not been done yet or it was completed exceptionally
+ if (!awaitFuture.isDone() || awaitFuture.isCompletedExceptionally()) {
+ awaitFuture
+ .whenComplete((ignored, ex) -> {
+ if (ex != null) {
+ // handles exception from asynchronous part of phase evaluation
+ onError(ex);
+ }
+ })
+ .thenRunAsync(this, executor::execute);
+
+ return;
+ }
+ }
+ } while (true);
+ }
+
+ CompletableFuture<Void> onPhaseStarted(ExecutionPhase phase) {
+ return onPhaseStartedCallback.computeIfAbsent(phase, k -> new CompletableFuture<>());
+ }
+
+ /** Moves the query to a given state. */
+ void moveTo(ExecutionPhase newPhase) {
+ synchronized (mux) {
+ assert currentPhase.transitionAllowed(newPhase) : "currentPhase=" + currentPhase + ", newPhase=" + newPhase;
+
+ currentPhase = newPhase;
+ }
+
+ onPhaseStartedCallback.computeIfAbsent(newPhase, k -> new CompletableFuture<>()).complete(null);
+ }
+
+ ExecutionPhase currentPhase() {
+ return currentPhase;
+ }
+
+ void onError(Throwable th) {
+ onPhaseStartedCallback.values().forEach(f -> f.completeExceptionally(th));
+
+ moveTo(ExecutionPhase.TERMINATED);
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java
new file mode 100644
index 0000000000..c955914834
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryExecutor.java
@@ -0,0 +1,342 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.stream.Collectors;
+import org.apache.ignite.internal.catalog.Catalog;
+import org.apache.ignite.internal.catalog.CatalogService;
+import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.schema.SchemaSyncService;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.sql.engine.InternalSqlRow;
+import org.apache.ignite.internal.sql.engine.QueryCancelledException;
+import org.apache.ignite.internal.sql.engine.QueryProperty;
+import org.apache.ignite.internal.sql.engine.SqlOperationContext;
+import org.apache.ignite.internal.sql.engine.exec.AsyncDataCursor;
+import org.apache.ignite.internal.sql.engine.exec.ExecutionService;
+import org.apache.ignite.internal.sql.engine.exec.LifecycleAware;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
+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.MultiStepPlan;
+import org.apache.ignite.internal.sql.engine.prepare.PrepareService;
+import org.apache.ignite.internal.sql.engine.prepare.QueryPlan;
+import org.apache.ignite.internal.sql.engine.property.SqlProperties;
+import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.sql.ParserService;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
+import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
+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.util.IgniteSpinBusyLock;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Executor which accepts requests for query execution and returns cursor to the result of execution.
+ */
+public class QueryExecutor implements LifecycleAware {
+ private final Cache<String, ParsedResult> queryToParsedResultCache;
+ private final ParserService parserService;
+ private final Executor executor;
+ private final ScheduledExecutorService scheduler;
+ private final ClockService clockService;
+ private final SchemaSyncService schemaSyncService;
+ private final PrepareService prepareService;
+ private final CatalogService catalogService;
+ private final ExecutionService executionService;
+ private final SqlProperties defaultProperties;
+ private final TransactionTracker transactionTracker;
+
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ private final ConcurrentMap<UUID, Query> runningQueries = new ConcurrentHashMap<>();
+
+ /**
+ * Creates executor.
+ *
+ * @param cacheFactory Factory to create cache for parsed AST.
+ * @param parsedResultsCacheSize Size of the cache for parsed AST.
+ * @param parserService Service to parse query string.
+ * @param executor Executor to submit query execution tasks.
+ * @param scheduler Scheduler to use for timeout task scheduling.
+ * @param clockService Current time provider.
+ * @param schemaSyncService Synchronization service to wait for metadata availability.
+ * @param prepareService Service to submit optimization
+ * @param catalogService Catalog service.
+ * @param executionService Service to submit query plans for execution.
+ * @param defaultProperties Set of properties to use as defaults.
+ * @param transactionTracker Tracker to track usage of transactions by query.
+ */
+ public QueryExecutor(
+ CacheFactory cacheFactory,
+ int parsedResultsCacheSize,
+ ParserService parserService,
+ Executor executor,
+ ScheduledExecutorService scheduler,
+ ClockService clockService,
+ SchemaSyncService schemaSyncService,
+ PrepareService prepareService,
+ CatalogService catalogService,
+ ExecutionService executionService,
+ SqlProperties defaultProperties,
+ TransactionTracker transactionTracker
+ ) {
+ this.queryToParsedResultCache = cacheFactory.create(parsedResultsCacheSize);
+ this.parserService = parserService;
+ this.executor = executor;
+ this.scheduler = scheduler;
+ this.clockService = clockService;
+ this.schemaSyncService = schemaSyncService;
+ this.prepareService = prepareService;
+ this.catalogService = catalogService;
+ this.executionService = executionService;
+ this.defaultProperties = defaultProperties;
+ this.transactionTracker = transactionTracker;
+ }
+
+ /**
+ * Executes the given query with provided parameters.
+ *
+ * <p>This is a common entry point for both single statement and script execution.
+ *
+ * @param properties User query properties. See {@link QueryProperty} for available properties.
+ * @param txContext Transactional context to use.
+ * @param sql Query string.
+ * @param params Query parameters.
+ * @return Future which will be completed with cursor.
+ */
+ public CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executeQuery(
+ SqlProperties properties,
+ QueryTransactionContext txContext,
+ String sql,
+ Object... params
+ ) {
+ SqlProperties properties0 = SqlPropertiesHelper.chain(properties, defaultProperties);
+
+ Query query = new Query(
+ Instant.ofEpochMilli(clockService.now().getPhysical()),
+ this,
+ UUID.randomUUID(),
+ sql,
+ properties0,
+ txContext,
+ params,
+ null
+ );
+
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ trackQuery(query);
+ } finally {
+ busyLock.leaveBusy();
+ }
+
+ long queryTimeout = properties.getOrDefault(QueryProperty.QUERY_TIMEOUT, 0L);
+
+ if (queryTimeout > 0) {
+ query.cancel.setTimeout(scheduler, queryTimeout);
+ }
+
+ // next state after CURSOR_INITIALIZATION. Cursor must be ready at this point.
+ CompletableFuture<Void> trigger = query.onPhaseStarted(ExecutionPhase.EXECUTING);
+
+ query.run();
+
+ return trigger
+ .thenApply(ignored -> query.cursor);
+ }
+
+ CompletableFuture<AsyncSqlCursor<InternalSqlRow>> executeChildQuery(
+ Query parent,
+ QueryTransactionContext scriptTxContext,
+ int statementNum,
+ ParsedResult parsedQuery,
+ Object[] params,
+ @Nullable CompletableFuture<AsyncSqlCursor<InternalSqlRow>> nextCursorFuture
+ ) {
+ Query query = new Query(
+ Instant.ofEpochMilli(clockService.now().getPhysical()),
+ parent,
+ parsedQuery,
+ statementNum,
+ UUID.randomUUID(),
+ scriptTxContext,
+ params,
+ nextCursorFuture
+ );
+
+ if (!busyLock.enterBusy()) {
+ return CompletableFuture.failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ trackQuery(query);
+ } finally {
+ busyLock.leaveBusy();
+ }
+
+ try {
+ parent.cancel.attach(query.cancel);
+ } catch (QueryCancelledException ex) {
+ return CompletableFuture.failedFuture(ex);
+ }
+
+ // next state after CURSOR_INITIALIZATION. Cursor must be ready at this point.
+ CompletableFuture<Void> trigger = query.onPhaseStarted(ExecutionPhase.EXECUTING);
+
+ // Child query already has parsed AST, therefore it's safe to transfer it directly to OPTIMIZATION phase.
+ query.moveTo(ExecutionPhase.OPTIMIZING);
+
+ query.run();
+
+ return trigger
+ .thenApply(ignored -> query.cursor);
+ }
+
+ /** Looks up parsed result in cache by given query string. */
+ public @Nullable ParsedResult lookupParsedResultInCache(String sql) {
+ return queryToParsedResultCache.get(sql);
+ }
+
+ /** Stores the given parsed statements in cache. */
+ public void updateParsedResultCache(String sql, ParsedResult result) {
+ queryToParsedResultCache.put(sql, result);
+ }
+
+ /** Parses the given query string. */
+ public ParsedResult parse(String sql) {
+ return parserService.parse(sql);
+ }
+
+ List<ParsedResult> parseScript(String sql) {
+ return parserService.parseScript(sql);
+ }
+
+ void execute(Runnable runnable) {
+ executor.execute(runnable);
+ }
+
+ HybridTimestamp deriveOperationTime(QueryTransactionContext txContext) {
+ QueryTransactionWrapper txWrapper = txContext.explicitTx();
+
+ if (txWrapper == null) {
+ return clockService.now();
+ }
+
+ return txWrapper.unwrap().startTimestamp();
+ }
+
+ CompletableFuture<Void> waitForMetadata(HybridTimestamp timestamp) {
+ return schemaSyncService.waitForMetadataCompleteness(timestamp);
+ }
+
+ CompletableFuture<QueryPlan> prepare(ParsedResult result, SqlOperationContext operationContext) {
+ return prepareService.prepareAsync(result, operationContext);
+ }
+
+ HybridTimestamp deriveMinimalRequiredTime(QueryPlan plan) {
+ Integer catalogVersion = null;
+
+ if (plan instanceof MultiStepPlan) {
+ catalogVersion = ((MultiStepPlan) plan).catalogVersion();
+ } else if (plan instanceof KeyValueModifyPlan) {
+ catalogVersion = ((KeyValueModifyPlan) plan).catalogVersion();
+ } else if (plan instanceof KeyValueGetPlan) {
+ catalogVersion = ((KeyValueGetPlan) plan).catalogVersion();
+ }
+
+ if (catalogVersion != null) {
+ Catalog catalog = catalogService.catalog(catalogVersion);
+
+ assert catalog != null;
+
+ return HybridTimestamp.hybridTimestamp(catalog.time());
+ }
+
+ return clockService.now();
+ }
+
+ AsyncDataCursor<InternalSqlRow> executePlan(
+ SqlOperationContext ctx,
+ QueryPlan plan
+ ) {
+ return executionService.executePlan(plan, ctx);
+ }
+
+ HybridTimestamp clockNow() {
+ return clockService.now();
+ }
+
+ MultiStatementHandler createScriptHandler(Query query) {
+ List<ParsedResult> parsedResults = query.parsedScript;
+
+ assert parsedResults != null;
+
+ return new MultiStatementHandler(
+ transactionTracker,
+ query,
+ query.txContext,
+ parsedResults,
+ query.params
+ );
+ }
+
+ private void trackQuery(Query query) {
+ Query old = runningQueries.put(query.id, query);
+
+ assert old == null : "Query with the same id already registered";
+
+ query.onPhaseStarted(ExecutionPhase.TERMINATED)
+ .whenComplete((ignored, ex) -> runningQueries.remove(query.id));
+ }
+
+ /** Returns list of queries registered on server at the moment. */
+ public List<QueryInfo> runningQueries() {
+ return runningQueries.values().stream()
+ .map(QueryInfo::new)
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public void start() {
+ // no-op
+ }
+
+ @Override
+ public void stop() throws Exception {
+ busyLock.block();
+
+ Exception ex = new NodeStoppingException();
+
+ runningQueries.values().forEach(query -> query.onError(ex));
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryInfo.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryInfo.java
new file mode 100644
index 0000000000..1cde919b5b
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/QueryInfo.java
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.time.Instant;
+import java.util.UUID;
+import org.apache.ignite.internal.sql.engine.QueryProperty;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.jetbrains.annotations.Nullable;
+
+/** Read-only snapshot of a query state. */
+public class QueryInfo {
+ private final UUID id;
+ private final @Nullable UUID parentId;
+ private final String schema;
+ private final String sql;
+ private final Instant startTime;
+ private final ExecutionPhase phase;
+ private final @Nullable SqlQueryType queryType;
+ private final @Nullable UUID transactionId;
+ private final int statementNum;
+ private final boolean script;
+
+ QueryInfo(Query query) {
+ id = query.id;
+ parentId = query.parentId;
+ statementNum = query.statementNum;
+ schema = query.properties.get(QueryProperty.DEFAULT_SCHEMA);
+ sql = query.sql;
+ startTime = query.createdAt;
+ phase = query.currentPhase();
+ queryType = deriveQueryType(query.parsedResult);
+ transactionId = deriveTxId(query);
+
+ script = query.parsedScript != null;
+ }
+
+ private static @Nullable SqlQueryType deriveQueryType(@Nullable ParsedResult parsedResult) {
+ return parsedResult == null ? null : parsedResult.queryType();
+ }
+
+ private static @Nullable UUID deriveTxId(Query query) {
+ InternalTransaction tx = query.usedTransaction;
+
+ return tx != null ? tx.id() : null;
+ }
+
+ /** Returns {@code true} if current query is a script, returns {@code false} otherwise. */
+ public boolean script() {
+ return script;
+ }
+
+ /** Returns id of the query. */
+ public UUID id() {
+ return id;
+ }
+
+ /** Returns id of the parent query, if any. */
+ public @Nullable UUID parentId() {
+ return parentId;
+ }
+
+ /** Returns name of the schema that was used to resolve non qualified object names. */
+ public String schema() {
+ return schema;
+ }
+
+ /** Returns an original query string. */
+ public String sql() {
+ return sql;
+ }
+
+ /** Returns time at which query appears on server. */
+ public Instant startTime() {
+ return startTime;
+ }
+
+ /** Returns current phase of query execution. */
+ public ExecutionPhase phase() {
+ return phase;
+ }
+
+ /** Returns type of the query, if known. */
+ public @Nullable SqlQueryType queryType() {
+ return queryType;
+ }
+
+ /** Returns id of the transaction, if known. */
+ public @Nullable UUID transactionId() {
+ return transactionId;
+ }
+
+ /** Returns 0-based index of query within the script, if applicable, returns {@code -1} otherwise. */
+ public int statementNum() {
+ return statementNum;
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/RegisteredPhaseHandler.java
similarity index 64%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/RegisteredPhaseHandler.java
index ebc857bf9f..d466d3b3dc 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/RegisteredPhaseHandler.java
@@ -15,24 +15,18 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine.exec.fsm;
-/**
- * Phases of query execution.
- */
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
+/** Handler that kick starts query processing. */
+class RegisteredPhaseHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new RegisteredPhaseHandler();
+
+ private RegisteredPhaseHandler() { }
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
+ @Override
+ public Result handle(Query query) {
+ query.moveTo(ExecutionPhase.PARSING);
- /**
- * Query execution.
- */
- EXECUTION
+ return Result.proceedImmediately();
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Result.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Result.java
new file mode 100644
index 0000000000..cdbc61033c
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/Result.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.tostring.IgniteToStringExclude;
+import org.apache.ignite.internal.tostring.S;
+import org.jetbrains.annotations.Nullable;
+
+/** Result of evaluation of particular {@link ExecutionPhase}. */
+class Result {
+ enum Status {
+ PROCEED_IMMEDIATELY, SCHEDULE, STOP
+ }
+
+ private final Status status;
+
+ @IgniteToStringExclude
+ private final @Nullable CompletableFuture<Void> await;
+
+ static Result proceedImmediately() {
+ return new Result(Status.PROCEED_IMMEDIATELY, null);
+ }
+
+ static Result proceedAfter(CompletableFuture<Void> stage) {
+ assert stage != null;
+
+ return new Result(Status.SCHEDULE, stage);
+ }
+
+ static Result stop() {
+ return new Result(Status.STOP, null);
+ }
+
+ private Result(Status status, @Nullable CompletableFuture<Void> await) {
+ this.status = status;
+ this.await = await;
+ }
+
+ Status status() {
+ return status;
+ }
+
+ @Nullable CompletableFuture<Void> await() {
+ return await;
+ }
+
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ScriptInitializationPhaseHandler.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ScriptInitializationPhaseHandler.java
new file mode 100644
index 0000000000..6216a92cce
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ScriptInitializationPhaseHandler.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+
+/** Initializes script handler and kick starts script execution. */
+class ScriptInitializationPhaseHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new ScriptInitializationPhaseHandler();
+
+ private ScriptInitializationPhaseHandler() {
+ }
+
+ @Override
+ public Result handle(Query query) {
+ List<ParsedResult> parsedResults = query.parsedScript;
+
+ assert parsedResults != null;
+
+ CompletableFuture<Void> awaitFuture = query.executor.createScriptHandler(query)
+ .processNext()
+ .thenAccept(cursor -> {
+ query.cursor = cursor;
+
+ query.moveTo(ExecutionPhase.EXECUTING);
+ });
+
+ return Result.proceedAfter(awaitFuture);
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/StopHandler.java
similarity index 68%
copy from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
copy to modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/StopHandler.java
index ebc857bf9f..e13998fc7a 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/StopHandler.java
@@ -15,24 +15,16 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine.exec.fsm;
-/**
- * Phases of query execution.
- */
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
+/** No-op handler that causes fsm to stop after current phase. */
+class StopHandler implements ExecutionPhaseHandler {
+ static final ExecutionPhaseHandler INSTANCE = new StopHandler();
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
+ private StopHandler() { }
- /**
- * Query execution.
- */
- EXECUTION
+ @Override
+ public Result handle(Query query) {
+ return Result.stop();
+ }
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ValidationHelper.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ValidationHelper.java
new file mode 100644
index 0000000000..63a85ab8e5
--- /dev/null
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/fsm/ValidationHelper.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.sql.engine.exec.fsm;
+
+import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.lang.ErrorGroups.Sql.STMT_VALIDATION_ERR;
+
+import java.util.Set;
+import org.apache.ignite.internal.sql.engine.QueryProperty;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.property.SqlProperties;
+import org.apache.ignite.internal.sql.engine.sql.ParsedResult;
+import org.apache.ignite.internal.sql.engine.util.TypeUtils;
+import org.apache.ignite.sql.SqlException;
+
+/**
+ * Helper class to share validation amongst different phase handlers as well as outside of the package.
+ */
+public final class ValidationHelper {
+ /** Performs additional validation of a parsed statement. **/
+ public static void validateParsedStatement(
+ SqlProperties properties,
+ ParsedResult parsedResult
+ ) {
+ Set<SqlQueryType> allowedTypes = properties.get(QueryProperty.ALLOWED_QUERY_TYPES);
+ SqlQueryType queryType = parsedResult.queryType();
+
+ if (parsedResult.queryType() == SqlQueryType.TX_CONTROL) {
+ String message = "Transaction control statement can not be executed as an independent statement";
+
+ throw new SqlException(STMT_VALIDATION_ERR, message);
+ }
+
+ if (!allowedTypes.contains(queryType)) {
+ String message = format("Invalid SQL statement type. Expected {} but got {}", allowedTypes, queryType);
+
+ throw new SqlException(STMT_VALIDATION_ERR, message);
+ }
+ }
+
+ /** Performs validation of dynamic params provided. **/
+ public static void validateDynamicParameters(int expectedParamsCount, Object[] params, boolean exactMatch) throws SqlException {
+ if (exactMatch && expectedParamsCount != params.length || params.length > expectedParamsCount) {
+ String message = format(
+ "Unexpected number of query parameters. Provided {} but there is only {} dynamic parameter(s).",
+ params.length, expectedParamsCount
+ );
+
+ throw new SqlException(STMT_VALIDATION_ERR, message);
+ }
+
+ for (Object param : params) {
+ if (!TypeUtils.supportParamInstance(param)) {
+ String message = format(
+ "Unsupported dynamic parameter defined. Provided '{}' is not supported.", param.getClass().getName());
+
+ throw new SqlException(STMT_VALIDATION_ERR, message);
+ }
+ }
+ }
+}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistry.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistry.java
deleted file mode 100644
index a1da133443..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistry.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import java.util.Collection;
-import java.util.UUID;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
-import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.apache.ignite.internal.systemview.api.SystemView;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Registry that keeps track of running queries and open cursors.
- */
-public interface RunningQueriesRegistry {
- /** Returns number of opened cursors. */
- int openedCursorsCount();
-
- /** Returns list of running queries. */
- Collection<RunningQueryInfo> queries();
-
- /** Registers a new query. */
- RunningQueryInfo registerQuery(String schema, String sql, @Nullable QueryTransactionWrapper txWrapper);
-
- /** Registers a new multi-statement query. */
- RunningScriptInfoTracker registerScript(String schema, String sql, ScriptTransactionContext scriptTxContext);
-
- /** Associates a new cursor with the query. */
- void registerCursor(RunningQueryInfo queryInfo, AsyncSqlCursor<?> cursor);
-
- /** Unregisters the query. */
- void unregister(UUID uuid);
-
- /** Returns a system view for running queries. */
- SystemView<RunningQueryInfo> asSystemView();
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryImpl.java
deleted file mode 100644
index 7405c5da17..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryImpl.java
+++ /dev/null
@@ -1,142 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import static org.apache.ignite.internal.type.NativeTypes.stringOf;
-
-import java.time.Instant;
-import java.util.Collection;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Flow.Publisher;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
-import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.apache.ignite.internal.systemview.api.SystemView;
-import org.apache.ignite.internal.systemview.api.SystemViews;
-import org.apache.ignite.internal.type.NativeType;
-import org.apache.ignite.internal.type.NativeTypes;
-import org.apache.ignite.internal.util.SubscriptionUtils;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Implementation of {@link RunningQueriesRegistry}.
- */
-public class RunningQueriesRegistryImpl implements RunningQueriesRegistry {
- public static final String SCRIPT_QUERY_TYPE = "SCRIPT";
-
- private final Map<UUID, RunningQueryInfo> runningQueries = new ConcurrentHashMap<>();
-
- private final AtomicInteger openedCursorsCount = new AtomicInteger();
-
- @Override
- public int openedCursorsCount() {
- return openedCursorsCount.get();
- }
-
- @Override
- public Collection<RunningQueryInfo> queries() {
- return runningQueries.values();
- }
-
- @Override
- public RunningQueryInfo registerQuery(String schema, String sql, @Nullable QueryTransactionWrapper txWrapper) {
- String explicitTxId = txWrapper != null ? txWrapper.unwrap().id().toString() : null;
-
- return putQueryInfo(UUID.randomUUID(), schema, sql, null, explicitTxId);
- }
-
- @Override
- public RunningScriptInfoTracker registerScript(String schema, String sql, ScriptTransactionContext scriptTxContext) {
- UUID queryId = UUID.randomUUID();
- QueryTransactionWrapper txWrapper = scriptTxContext.explicitTx();
- String txId = txWrapper != null ? txWrapper.unwrap().id().toString() : null;
-
- RunningQueryInfoImpl queryInfo = putQueryInfo(queryId, schema, sql, SCRIPT_QUERY_TYPE, txId);
-
- return new RunningScriptInfoTrackerImpl(queryInfo, scriptTxContext, runningQueries);
- }
-
- @Override
- public void registerCursor(RunningQueryInfo queryInfo, AsyncSqlCursor<?> cursor) {
- openedCursorsCount.incrementAndGet();
-
- queryInfo.cursor(cursor);
-
- cursor.onClose().whenComplete((r, e) -> unregister(queryInfo.queryId()));
- }
-
- @Override
- public void unregister(UUID uuid) {
- RunningQueryInfo info = runningQueries.remove(uuid);
-
- if (info != null) {
- if (info.cursor() != null) {
- openedCursorsCount.decrementAndGet();
- }
-
- RunningScriptInfoTracker script = ((RunningQueryInfoImpl) info).script();
-
- if (script != null) {
- script.onStatementUnregistered();
- }
- }
- }
-
- @Override
- public SystemView<RunningQueryInfo> asSystemView() {
- Publisher<RunningQueryInfo> viewDataPublisher = SubscriptionUtils.fromIterable(() -> queries().iterator());
-
- NativeType stringType = stringOf(Short.MAX_VALUE);
- NativeType idType = stringOf(36);
-
- NativeType timestampType = NativeTypes.timestamp(NativeTypes.MAX_TIME_PRECISION);
-
- return SystemViews.<RunningQueryInfo>nodeViewBuilder()
- .name("SQL_QUERIES")
- .nodeNameColumnAlias("INITIATOR_NODE")
- .<String>addColumn("ID", idType, info -> info.queryId().toString())
- .<String>addColumn("PHASE", stringOf(10), RunningQueryInfo::phase)
- .<String>addColumn("TYPE", stringOf(10), RunningQueryInfo::queryType)
- .<String>addColumn("SCHEMA", stringType, RunningQueryInfo::schema)
- .<String>addColumn("SQL", stringType, RunningQueryInfo::sql)
- .<Instant>addColumn("START_TIME", timestampType, RunningQueryInfo::startTime)
- .<String>addColumn("TRANSACTION_ID", idType, RunningQueryInfo::transactionId)
- .<String>addColumn("PARENT_ID", idType, RunningQueryInfo::parentId)
- .<Integer>addColumn("STATEMENT_NUM", NativeTypes.INT32, RunningQueryInfo::statementNumber)
- .dataProvider(viewDataPublisher)
- .build();
- }
-
- private RunningQueryInfoImpl putQueryInfo(
- UUID queryId,
- String schema,
- String sql,
- @Nullable String type,
- @Nullable String transactionId
- ) {
- RunningQueryInfoImpl queryInfo =
- new RunningQueryInfoImpl(queryId, schema, sql, Instant.now(), transactionId, type, null, null);
-
- runningQueries.put(queryId, queryInfo);
-
- return queryInfo;
- }
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfo.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfo.java
deleted file mode 100644
index b2fe4c7fc4..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfo.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import java.time.Instant;
-import java.util.UUID;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Information about the running query, which is exposed to the {@code SQL_QUERIES} system view.
- *
- * <p>Some fields are updated during query execution and the information in them may not be available
- * until a particular execution phase occurs.
- */
-public interface RunningQueryInfo {
- /** Returns the query identifier. */
- UUID queryId();
-
- /** Returns the schema name. */
- String schema();
-
- /** Returns the original SQL query text. */
- String sql();
-
- /** Returns the query start time. */
- Instant startTime();
-
- /** Returns the query execution phase. */
- String phase();
-
- /** Sets the query execution phase. */
- void phase(QueryExecutionPhase phase);
-
- /** Returns the query type, or {@code null} if the query type is not yet known. */
- @Nullable String queryType();
-
- /** Sets the query type. */
- void queryType(SqlQueryType queryType);
-
- /** Returns the transaction ID or {@code null} if the transaction has not been started. */
- @Nullable String transactionId();
-
- /** Sets the transaction identifier. */
- void transactionId(UUID txId);
-
- /** Returns the cursor associated with the query or {@code null} if no cursor was opened. */
- @Nullable AsyncSqlCursor<?> cursor();
-
- /** Sets the open cursor associated with the query. */
- void cursor(AsyncSqlCursor<?> cursor);
-
- /** Returns the ID of the parent multi-statement query or {@code null} if the query is not a part of multi-statement query. */
- @Nullable String parentId();
-
- /** Returns the statement number in the multi-statement query or {@code null} if the query is not a part of multi-statement query. */
- @Nullable Integer statementNumber();
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfoImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfoImpl.java
deleted file mode 100644
index 45004e92c4..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningQueryInfoImpl.java
+++ /dev/null
@@ -1,148 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import java.time.Instant;
-import java.util.Objects;
-import java.util.UUID;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Information about a running query.
- */
-public class RunningQueryInfoImpl implements RunningQueryInfo {
- private final UUID queryId;
- private final String schema;
- private final String sql;
- private final Instant startTime;
- private final Integer statementNumber;
- private final @Nullable String parentId;
- private final @Nullable RunningScriptInfoTracker script;
-
- private volatile @Nullable String txId;
- private volatile String phase;
- private volatile @Nullable String queryType;
- private volatile @Nullable AsyncSqlCursor<?> cursor;
-
- /**
- * Constructor.
- */
- public RunningQueryInfoImpl(
- UUID queryId,
- String schema,
- String sql,
- Instant startTime,
- @Nullable String transactionId,
- @Nullable String queryType,
- @Nullable Integer statementNumber,
- @Nullable RunningScriptInfoTracker script
- ) {
- this.queryId = queryId;
- this.schema = schema;
- this.sql = sql;
- this.txId = transactionId;
- this.script = script;
- this.parentId = script == null ? null : script.queryId().toString();
- this.queryType = queryType;
- this.startTime = startTime;
- this.statementNumber = statementNumber;
- this.phase = QueryExecutionPhase.INITIALIZATION.name();
- }
-
- @Override
- public UUID queryId() {
- return queryId;
- }
-
- @Override
- public String schema() {
- return schema;
- }
-
- @Override
- public String sql() {
- return sql;
- }
-
- @Override
- public Instant startTime() {
- return startTime;
- }
-
- @Override
- public String phase() {
- return phase;
- }
-
- @Override
- public void phase(QueryExecutionPhase phase) {
- this.phase = phase.name();
- }
-
- @Override
- public @Nullable String queryType() {
- return queryType;
- }
-
- @Override
- public void queryType(SqlQueryType queryType) {
- this.queryType = queryType.name();
- }
-
- @Override
- public @Nullable String transactionId() {
- return txId;
- }
-
- @Override
- public void transactionId(UUID id) {
- assert txId == null;
-
- txId = id.toString();
- }
-
- @Override
- public @Nullable AsyncSqlCursor<?> cursor() {
- return cursor;
- }
-
- @Override
- public void cursor(AsyncSqlCursor<?> cursor) {
- Objects.requireNonNull(cursor, "cursor");
-
- assert this.cursor == null;
-
- this.cursor = cursor;
- }
-
- @Override
- public @Nullable String parentId() {
- return parentId;
- }
-
- @Override
- public @Nullable Integer statementNumber() {
- return statementNumber;
- }
-
- @Nullable RunningScriptInfoTracker script() {
- return script;
- }
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTracker.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTracker.java
deleted file mode 100644
index 4d337384e2..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTracker.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import java.util.UUID;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-
-/**
- * Manages information about the execution of a multi-statement query.
- *
- * <ul>
- * <li>The script has two phases: {@link QueryExecutionPhase#INITIALIZATION initialization}
- * and {@link QueryExecutionPhase#EXECUTION execution}.</li>
- * <li>The script unregisters itself in the {@link RunningQueriesRegistry} when the last
- * sub-statement unregisters itself.</li>
- * </ul>
- */
-public interface RunningScriptInfoTracker {
- /** Returns the identifier of multi-statement query. */
- UUID queryId();
-
- /** Registers a new statement in the {@link RunningQueriesRegistry}. */
- RunningQueryInfo registerStatement(String sql, SqlQueryType queryType);
-
- /** Must be called when sub-statement unregisters from the {@link RunningQueriesRegistry}. */
- void onStatementUnregistered();
-
- /** Sets remaining sub-statements count and switches script execution phase to {@link QueryExecutionPhase#EXECUTION}. */
- void onInitializationComplete(int statementsCount);
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTrackerImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTrackerImpl.java
deleted file mode 100644
index 2e2c7a1c2d..0000000000
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/RunningScriptInfoTrackerImpl.java
+++ /dev/null
@@ -1,103 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import java.time.Instant;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
-import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Implementation of {@link RunningScriptInfoTracker}.
- */
-class RunningScriptInfoTrackerImpl implements RunningScriptInfoTracker {
- private final AtomicInteger registeredStatementsCounter = new AtomicInteger();
-
- private final AtomicInteger remainingStatementsCounter = new AtomicInteger();
-
- private final Map<UUID, RunningQueryInfo> runningQueries;
-
- private final RunningQueryInfoImpl scriptInfo;
-
- private final ScriptTransactionContext scriptTxCtx;
-
- RunningScriptInfoTrackerImpl(
- RunningQueryInfoImpl queryInfo,
- ScriptTransactionContext scriptTxCtx,
- Map<UUID, RunningQueryInfo> runningQueries
- ) {
- this.scriptInfo = queryInfo;
- this.scriptTxCtx = scriptTxCtx;
- this.runningQueries = runningQueries;
- }
-
- @Override
- public UUID queryId() {
- return scriptInfo.queryId();
- }
-
- @Override
- public @Nullable RunningQueryInfo registerStatement(String sql, SqlQueryType queryType) {
- remainingStatementsCounter.decrementAndGet();
-
- if (queryType == SqlQueryType.TX_CONTROL) {
- // We just need to update count of processed statements.
- return null;
- }
-
- UUID queryId = UUID.randomUUID();
-
- // This can be an explicit transaction passed as a parameter
- // or started using the tx control script statement.
- QueryTransactionWrapper txWrapper = scriptTxCtx.explicitTx();
-
- RunningQueryInfoImpl statementQueryInfo = new RunningQueryInfoImpl(
- queryId,
- scriptInfo.schema(),
- sql,
- Instant.now(),
- txWrapper == null ? null : txWrapper.unwrap().id().toString(),
- queryType.name(),
- registeredStatementsCounter.incrementAndGet(),
- this
- );
-
- runningQueries.put(queryId, statementQueryInfo);
-
- return statementQueryInfo;
- }
-
- @Override
- public void onStatementUnregistered() {
- int remainingCount = registeredStatementsCounter.decrementAndGet();
-
- if (remainingCount == 0 && remainingStatementsCounter.get() == 0) {
- runningQueries.remove(scriptInfo.queryId());
- }
- }
-
- @Override
- public void onInitializationComplete(int statementsCount) {
- scriptInfo.phase(QueryExecutionPhase.EXECUTION);
- remainingStatementsCounter.set(statementsCount);
- }
-}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
index b1c12873f8..ab1aefae27 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
@@ -21,10 +21,10 @@ import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_ERR;
import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.tx.TransactionException;
import org.jetbrains.annotations.Nullable;
@@ -35,19 +35,19 @@ public class QueryTransactionContextImpl implements QueryTransactionContext {
private final TxManager txManager;
private final HybridTimestampTracker observableTimeTracker;
private final @Nullable QueryTransactionWrapper tx;
- private final TransactionInflights transactionInflights;
+ private final TransactionTracker txTracker;
/** Constructor. */
public QueryTransactionContextImpl(
TxManager txManager,
HybridTimestampTracker observableTimeTracker,
@Nullable InternalTransaction tx,
- TransactionInflights transactionInflights
+ TransactionTracker txTracker
) {
this.txManager = txManager;
this.observableTimeTracker = observableTimeTracker;
- this.tx = tx != null ? new QueryTransactionWrapperImpl(tx, false, transactionInflights) : null;
- this.transactionInflights = transactionInflights;
+ this.tx = tx != null ? new QueryTransactionWrapperImpl(tx, false, txTracker) : null;
+ this.txTracker = txTracker;
}
/**
@@ -63,14 +63,14 @@ public class QueryTransactionContextImpl implements QueryTransactionContext {
if (tx == null) {
transaction = txManager.begin(observableTimeTracker, readOnly);
- result = new QueryTransactionWrapperImpl(transaction, true, transactionInflights);
+ result = new QueryTransactionWrapperImpl(transaction, true, txTracker);
} else {
transaction = tx.unwrap();
result = tx;
}
// Adding inflights only for read-only transactions. See TransactionInflights.ReadOnlyTxContext for details.
- if (transaction.isReadOnly() && !transactionInflights.addInflight(transaction.id(), transaction.isReadOnly())) {
+ if (transaction.isReadOnly() && !txTracker.register(transaction.id(), transaction.isReadOnly())) {
throw new TransactionException(TX_ALREADY_FINISHED_ERR, format("Transaction is already finished [tx={}]", transaction));
}
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionWrapperImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionWrapperImpl.java
index 01d626f55e..6f64668dea 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionWrapperImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionWrapperImpl.java
@@ -21,8 +21,8 @@ import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFu
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
/**
* Wrapper for the transaction that encapsulates the management of an implicit transaction.
@@ -32,7 +32,7 @@ public class QueryTransactionWrapperImpl implements QueryTransactionWrapper {
private final InternalTransaction transaction;
- private final TransactionInflights transactionInflights;
+ private final TransactionTracker txTracker;
private final AtomicBoolean committedImplicit = new AtomicBoolean();
@@ -41,12 +41,12 @@ public class QueryTransactionWrapperImpl implements QueryTransactionWrapper {
*
* @param transaction Transaction.
* @param implicit Whether tx is implicit.
- * @param transactionInflights Transaction inflights.
+ * @param txTracker Transaction tracker.
*/
- public QueryTransactionWrapperImpl(InternalTransaction transaction, boolean implicit, TransactionInflights transactionInflights) {
+ public QueryTransactionWrapperImpl(InternalTransaction transaction, boolean implicit, TransactionTracker txTracker) {
this.transaction = transaction;
this.implicit = implicit;
- this.transactionInflights = transactionInflights;
+ this.txTracker = txTracker;
}
@Override
@@ -57,7 +57,7 @@ public class QueryTransactionWrapperImpl implements QueryTransactionWrapper {
@Override
public CompletableFuture<Void> commitImplicit() {
if (transaction.isReadOnly() && committedImplicit.compareAndSet(false, true)) {
- transactionInflights.removeInflight(transaction.id());
+ txTracker.unregister(transaction.id());
}
if (implicit) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionContext.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionContext.java
index 1fedd2bc1d..6e158e4ea5 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionContext.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionContext.java
@@ -27,11 +27,11 @@ import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
import org.apache.ignite.internal.sql.engine.TxControlInsideExternalTxNotSupportedException;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.sql.engine.sql.IgniteSqlCommitTransaction;
import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransaction;
import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransactionMode;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.sql.SqlException;
import org.jetbrains.annotations.Nullable;
@@ -41,16 +41,16 @@ import org.jetbrains.annotations.Nullable;
public class ScriptTransactionContext implements QueryTransactionContext {
private final QueryTransactionContextImpl txContext;
- private final TransactionInflights transactionInflights;
+ private final TransactionTracker txTracker;
private volatile @Nullable ScriptTransactionWrapperImpl wrapper;
/** Constructor. */
- public ScriptTransactionContext(QueryTransactionContext txContext, TransactionInflights transactionInflights) {
+ public ScriptTransactionContext(QueryTransactionContext txContext, TransactionTracker txTracker) {
assert txContext instanceof QueryTransactionContextImpl : txContext;
this.txContext = (QueryTransactionContextImpl) txContext;
- this.transactionInflights = transactionInflights;
+ this.txTracker = txTracker;
}
/**
@@ -108,7 +108,7 @@ public class ScriptTransactionContext implements QueryTransactionContext {
boolean readOnly = ((IgniteSqlStartTransaction) node).getMode() == IgniteSqlStartTransactionMode.READ_ONLY;
InternalTransaction tx = txContext.getOrStartImplicit(readOnly).unwrap();
- this.wrapper = new ScriptTransactionWrapperImpl(tx, transactionInflights);
+ this.wrapper = new ScriptTransactionWrapperImpl(tx, txTracker);
return nullCompletedFuture();
} else {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
index 2252cd995a..0bf34a24aa 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/ScriptTransactionWrapperImpl.java
@@ -30,8 +30,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.internal.util.AsyncCursor;
import org.apache.ignite.sql.SqlException;
@@ -64,13 +64,13 @@ class ScriptTransactionWrapperImpl implements QueryTransactionWrapper {
private Throwable rollbackCause;
- private final TransactionInflights transactionInflights;
+ private final TransactionTracker txTracker;
private final AtomicBoolean completedTx = new AtomicBoolean();
- ScriptTransactionWrapperImpl(InternalTransaction managedTx, TransactionInflights transactionInflights) {
+ ScriptTransactionWrapperImpl(InternalTransaction managedTx, TransactionTracker txTracker) {
this.managedTx = managedTx;
- this.transactionInflights = transactionInflights;
+ this.txTracker = txTracker;
}
@Override
@@ -193,7 +193,7 @@ class ScriptTransactionWrapperImpl implements QueryTransactionWrapper {
}
if (managedTx.isReadOnly() && completedTx.compareAndSet(false, true)) {
- transactionInflights.removeInflight(managedTx.id());
+ txTracker.unregister(managedTx.id());
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/QueryTransactionWrapperSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/QueryTransactionWrapperSelfTest.java
index f12c5615cb..01fc28d143 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/QueryTransactionWrapperSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/QueryTransactionWrapperSelfTest.java
@@ -33,6 +33,7 @@ import static org.mockito.Mockito.when;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
import org.apache.ignite.internal.sql.engine.framework.NoOpTransaction;
import org.apache.ignite.internal.sql.engine.sql.IgniteSqlCommitTransaction;
import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransaction;
@@ -45,7 +46,6 @@ import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.TxManager;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.lang.ErrorGroups.Sql;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -64,16 +64,16 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
private TxManager txManager;
@Mock
- private TransactionInflights transactionInflights;
+ private TransactionTracker transactionTracker;
@Test
public void testImplicitTransactionAttributes() {
prepareTransactionsMocks();
- when(transactionInflights.addInflight(any(), anyBoolean())).thenAnswer(inv -> true);
+ when(transactionTracker.register(any(), anyBoolean())).thenAnswer(inv -> true);
QueryTransactionContext transactionHandler = new QueryTransactionContextImpl(txManager, observableTimeTracker, null,
- transactionInflights);
+ transactionTracker);
QueryTransactionWrapper transactionWrapper = transactionHandler.getOrStartImplicit(false);
assertThat(transactionWrapper.unwrap().isReadOnly(), equalTo(false));
@@ -86,7 +86,7 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
public void commitImplicitTxNotAffectExternalTransaction() {
NoOpTransaction externalTx = new NoOpTransaction("test");
- QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(externalTx, false, transactionInflights);
+ QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(externalTx, false, transactionTracker);
wrapper.commitImplicit();
assertFalse(externalTx.commitFuture().isDone());
}
@@ -94,7 +94,7 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
@Test
public void testCommitImplicit() {
NoOpTransaction tx = new NoOpTransaction("test");
- QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(tx, true, transactionInflights);
+ QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(tx, true, transactionTracker);
wrapper.commitImplicit();
@@ -105,7 +105,7 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
@Test
public void testRollbackImplicit() {
NoOpTransaction tx = new NoOpTransaction("test");
- QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(tx, true, transactionInflights);
+ QueryTransactionWrapperImpl wrapper = new QueryTransactionWrapperImpl(tx, true, transactionTracker);
wrapper.rollback(null);
@@ -116,8 +116,8 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
@Test
public void throwsExceptionForTxControlStatementInsideExternalTransaction() {
ScriptTransactionContext txCtx = new ScriptTransactionContext(
- new QueryTransactionContextImpl(txManager, observableTimeTracker, new NoOpTransaction("test"), transactionInflights),
- transactionInflights
+ new QueryTransactionContextImpl(txManager, observableTimeTracker, new NoOpTransaction("test"), transactionTracker),
+ transactionTracker
);
assertThrowsExactly(TxControlInsideExternalTxNotSupportedException.class, () -> txCtx.handleControlStatement(null));
@@ -126,8 +126,8 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
@Test
public void throwsExceptionForNestedScriptTransaction() {
ScriptTransactionContext txCtx = new ScriptTransactionContext(
- new QueryTransactionContextImpl(txManager, observableTimeTracker, null, transactionInflights),
- transactionInflights
+ new QueryTransactionContextImpl(txManager, observableTimeTracker, null, transactionTracker),
+ transactionTracker
);
IgniteSqlStartTransaction txStartStmt = mock(IgniteSqlStartTransaction.class);
@@ -152,13 +152,13 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
prepareTransactionsMocks();
QueryTransactionContext implicitDmlTxCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, null,
- transactionInflights);
+ transactionTracker);
implicitDmlTxCtx.getOrStartImplicit(false);
// Check that RW txns do not create tx inflights.
assertTrue(inflights.isEmpty());
QueryTransactionContext implicitQueryTxCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, null,
- transactionInflights);
+ transactionTracker);
QueryTransactionWrapper implicitQueryTxWrapper = implicitQueryTxCtx.getOrStartImplicit(true);
assertTrue(inflights.contains(implicitQueryTxWrapper.unwrap().id()));
implicitQueryTxWrapper.commitImplicit();
@@ -166,14 +166,14 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
NoOpTransaction rwTx = NoOpTransaction.readWrite("test-rw");
QueryTransactionContext explicitRwTxCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, rwTx,
- transactionInflights);
+ transactionTracker);
explicitRwTxCtx.getOrStartImplicit(true);
// Check that RW txns do not create tx inflights.
assertTrue(inflights.isEmpty());
NoOpTransaction roTx = NoOpTransaction.readOnly("test-ro");
QueryTransactionContext explicitRoTxCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, roTx,
- transactionInflights);
+ transactionTracker);
QueryTransactionWrapper explicitRoTxWrapper = explicitRoTxCtx.getOrStartImplicit(true);
assertTrue(inflights.contains(explicitRoTxWrapper.unwrap().id()));
explicitRoTxWrapper.commitImplicit();
@@ -188,8 +188,8 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
prepareTransactionsMocks();
- QueryTransactionContext txCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, null, transactionInflights);
- ScriptTransactionContext scriptRwTxCtx = new ScriptTransactionContext(txCtx, transactionInflights);
+ QueryTransactionContext txCtx = new QueryTransactionContextImpl(txManager, observableTimeTracker, null, transactionTracker);
+ ScriptTransactionContext scriptRwTxCtx = new ScriptTransactionContext(txCtx, transactionTracker);
IgniteSqlStartTransaction sqlStartRwTx = mock(IgniteSqlStartTransaction.class);
when(sqlStartRwTx.getMode()).thenAnswer(inv -> IgniteSqlStartTransactionMode.READ_WRITE);
@@ -197,7 +197,7 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
scriptRwTxCtx.handleControlStatement(sqlStartRwTx);
assertTrue(inflights.isEmpty());
- ScriptTransactionContext scriptRoTxCtx = new ScriptTransactionContext(txCtx, transactionInflights);
+ ScriptTransactionContext scriptRoTxCtx = new ScriptTransactionContext(txCtx, transactionTracker);
IgniteSqlStartTransaction sqlStartRoTx = mock(IgniteSqlStartTransaction.class);
when(sqlStartRoTx.getMode()).thenAnswer(inv -> IgniteSqlStartTransactionMode.READ_ONLY);
@@ -227,8 +227,8 @@ public class QueryTransactionWrapperSelfTest extends BaseIgniteAbstractTest {
}
private void prepareTxInflightsMocks(Set<UUID> inflights) {
- when(transactionInflights.addInflight(any(), anyBoolean())).thenAnswer(inv -> inflights.add(inv.getArgument(0)));
+ when(transactionTracker.register(any(), anyBoolean())).thenAnswer(inv -> inflights.add(inv.getArgument(0)));
- doAnswer(inv -> inflights.remove(inv.getArgument(0))).when(transactionInflights).removeInflight(any());
+ doAnswer(inv -> inflights.remove(inv.getArgument(0))).when(transactionTracker).unregister(any());
}
}
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 1ecbb997bd..75f8516704 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
@@ -941,7 +941,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
QueryPlan plan = prepare("SELECT * FROM test_tbl", planCtx);
QueryCancel queryCancel = new QueryCancel();
- CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, 0);
+ CompletableFuture<Void> timeoutFut = setTimeout(queryCancel, 0);
SqlOperationContext ctx = operationContext(null)
.cancel(queryCancel)
@@ -1068,7 +1068,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
QueryPlan plan = prepare("SELECT * FROM test_tbl", operationContext(null).build());
QueryCancel queryCancel = new QueryCancel();
- CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, 50);
+ CompletableFuture<Void> timeoutFut = setTimeout(queryCancel, 50);
SqlOperationContext ctx = operationContext(null)
.cancel(queryCancel)
@@ -1280,7 +1280,7 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
for (int k = 0; k < attempts; k++) {
QueryCancel queryCancel = new QueryCancel();
- CompletableFuture<Void> timeoutFut = queryCancel.setTimeout(scheduler, deadlineMillis);
+ CompletableFuture<Void> timeoutFut = setTimeout(queryCancel, deadlineMillis);
SqlOperationContext execCtx = execCtxFunc.apply(queryCancel);
@@ -1629,4 +1629,17 @@ public class ExecutionServiceImplTest extends BaseIgniteAbstractTest {
};
}
}
+
+ private CompletableFuture<Void> setTimeout(QueryCancel queryCancel, long millis) {
+ CompletableFuture<Void> timeoutFut = new CompletableFuture<>();
+ queryCancel.add(timeout -> {
+ if (timeout) {
+ timeoutFut.complete(null);
+ }
+ });
+
+ queryCancel.setTimeout(scheduler, millis);
+
+ return timeoutFut;
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
index 7889639379..0fee72c9f7 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/TxAwareCursorSelfTest.java
@@ -32,7 +32,6 @@ import org.apache.ignite.internal.sql.engine.framework.NoOpTransaction;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapperImpl;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.apache.ignite.internal.util.AsyncCursor;
import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
import org.apache.ignite.internal.util.AsyncWrapper;
@@ -120,6 +119,6 @@ public class TxAwareCursorSelfTest extends BaseIgniteAbstractTest {
}
private static QueryTransactionWrapper newTxWrapper(boolean implicit) {
- return new QueryTransactionWrapperImpl(NoOpTransaction.readOnly("TX"), implicit, Mockito.mock(TransactionInflights.class));
+ return new QueryTransactionWrapperImpl(NoOpTransaction.readOnly("TX"), implicit, Mockito.mock(TransactionTracker.class));
}
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ExplicitTxContext.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ExplicitTxContext.java
index 09b1881bdf..6dbde499dd 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ExplicitTxContext.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ExplicitTxContext.java
@@ -17,35 +17,21 @@
package org.apache.ignite.internal.sql.engine.framework;
-import java.util.UUID;
-import org.apache.ignite.internal.hlc.ClockServiceImpl;
-import org.apache.ignite.internal.hlc.ClockWaiter;
-import org.apache.ignite.internal.hlc.HybridClock;
-import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapperImpl;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.jetbrains.annotations.Nullable;
/** Context that always returns explicit transaction. */
public class ExplicitTxContext implements QueryTransactionContext {
- private static final HybridClock CLOCK = new HybridClockImpl();
-
- private static final TransactionInflights TX_INFLIGHTS = new TransactionInflights(
- new TestPlacementDriver("dummy", UUID.randomUUID()),
- new ClockServiceImpl(CLOCK, new ClockWaiter("dummy", CLOCK), () -> 1L)
- );
-
private final HybridTimestampTracker observableTimeTracker = new HybridTimestampTracker();
private final QueryTransactionWrapper txWrapper;
public static QueryTransactionContext fromTx(InternalTransaction tx) {
- return new ExplicitTxContext(new QueryTransactionWrapperImpl(tx, false, TX_INFLIGHTS));
+ return new ExplicitTxContext(new QueryTransactionWrapperImpl(tx, false, NoOpTransactionTracker.INSTANCE));
}
private ExplicitTxContext(QueryTransactionWrapper txWrapper) {
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ImplicitTxContext.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ImplicitTxContext.java
index b8ab08b60a..32c4048a2a 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ImplicitTxContext.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/ImplicitTxContext.java
@@ -17,29 +17,15 @@
package org.apache.ignite.internal.sql.engine.framework;
-import java.util.UUID;
-import org.apache.ignite.internal.hlc.ClockServiceImpl;
-import org.apache.ignite.internal.hlc.ClockWaiter;
-import org.apache.ignite.internal.hlc.HybridClock;
-import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.placementdriver.TestPlacementDriver;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapperImpl;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
import org.jetbrains.annotations.Nullable;
/** Context that always creates implicit transaction. */
public class ImplicitTxContext implements QueryTransactionContext {
- private static final HybridClock CLOCK = new HybridClockImpl();
-
- private static final TransactionInflights TX_INFLIGHTS = new TransactionInflights(
- new TestPlacementDriver("dummy", UUID.randomUUID()),
- new ClockServiceImpl(CLOCK, new ClockWaiter("dummy", CLOCK), () -> 1L)
- );
-
public static final QueryTransactionContext INSTANCE = new ImplicitTxContext();
private final HybridTimestampTracker observableTimeTracker = new HybridTimestampTracker();
@@ -48,7 +34,7 @@ public class ImplicitTxContext implements QueryTransactionContext {
@Override
public QueryTransactionWrapper getOrStartImplicit(boolean readOnly) {
- return new QueryTransactionWrapperImpl(new NoOpTransaction("dummy"), true, TX_INFLIGHTS);
+ return new QueryTransactionWrapperImpl(new NoOpTransaction("dummy"), true, NoOpTransactionTracker.INSTANCE);
}
@Override
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransactionTracker.java
similarity index 62%
rename from modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
rename to modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransactionTracker.java
index ebc857bf9f..de3b7ece40 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/registry/QueryExecutionPhase.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransactionTracker.java
@@ -15,24 +15,24 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.sql.engine.registry;
+package org.apache.ignite.internal.sql.engine.framework;
+
+import java.util.UUID;
+import org.apache.ignite.internal.sql.engine.exec.TransactionTracker;
/**
- * Phases of query execution.
+ * Dummy no-op tracker.
*/
-public enum QueryExecutionPhase {
- /**
- * Query parsing, metadata synchronization.
- */
- INITIALIZATION,
+final class NoOpTransactionTracker implements TransactionTracker {
+ static final TransactionTracker INSTANCE = new NoOpTransactionTracker();
+
+ @Override
+ public boolean register(UUID txId, boolean readOnly) {
+ return true;
+ }
- /**
- * Query validation and optimization.
- */
- OPTIMIZATION,
+ @Override
+ public void unregister(UUID txId) {
- /**
- * Query execution.
- */
- EXECUTION
+ }
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoopRunningQueryInfo.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoopRunningQueryInfo.java
deleted file mode 100644
index 72a1d02194..0000000000
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoopRunningQueryInfo.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.framework;
-
-import java.time.Instant;
-import java.util.UUID;
-import org.apache.ignite.internal.sql.SqlCommon;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.apache.ignite.internal.sql.engine.registry.QueryExecutionPhase;
-import org.apache.ignite.internal.sql.engine.registry.RunningQueryInfo;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * No-op implementation of {@link RunningQueryInfo}.
- */
-public class NoopRunningQueryInfo implements RunningQueryInfo {
- private final UUID queryId;
- private final Instant startTime;
-
- /**
- * Constructor.
- */
- NoopRunningQueryInfo(UUID queryId) {
- this.queryId = queryId;
- this.startTime = Instant.now();
- }
-
- @Override
- public UUID queryId() {
- return queryId;
- }
-
- @Override
- public String schema() {
- return SqlCommon.DEFAULT_SCHEMA_NAME;
- }
-
- @Override
- public String sql() {
- return "";
- }
-
- @Override
- public Instant startTime() {
- return startTime;
- }
-
- @Override
- public String phase() {
- return QueryExecutionPhase.INITIALIZATION.name();
- }
-
- @Override
- public void phase(QueryExecutionPhase phase) {
- // No-op.
- }
-
- @Override
- public @Nullable String queryType() {
- return null;
- }
-
- @Override
- public void queryType(SqlQueryType queryType) {
- // No-op.
- }
-
- @Override
- public @Nullable String transactionId() {
- return null;
- }
-
- @Override
- public void transactionId(UUID id) {
- // No-op.
- }
-
- @Override
- public @Nullable AsyncSqlCursor<?> cursor() {
- return null;
- }
-
- @Override
- public void cursor(AsyncSqlCursor<?> cursor) {
- // No-op.
- }
-
- @Override
- public @Nullable String parentId() {
- return null;
- }
-
- @Override
- public @Nullable Integer statementNumber() {
- return null;
- }
-}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
index b0cd335d88..e3d059e1cd 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/TestNode.java
@@ -327,7 +327,6 @@ public class TestNode implements LifecycleAware {
.timeZoneId(SqlQueryProcessor.DEFAULT_TIME_ZONE_ID)
.txContext(txContext)
.parameters(params)
- .queryInfo(new NoopRunningQueryInfo(queryId))
.prefetchCallback(new PrefetchCallback());
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryTest.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryTest.java
deleted file mode 100644
index a3856d3541..0000000000
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/registry/RunningQueriesRegistryTest.java
+++ /dev/null
@@ -1,224 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.sql.engine.registry;
-
-import static org.apache.ignite.internal.sql.engine.registry.RunningQueriesRegistryImpl.SCRIPT_QUERY_TYPE;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.lessThanOrEqualTo;
-import static org.hamcrest.Matchers.not;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.jupiter.api.Assertions.assertSame;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyBoolean;
-import static org.mockito.Mockito.when;
-
-import java.time.Instant;
-import java.util.concurrent.CompletableFuture;
-import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.apache.ignite.internal.sql.engine.framework.ExplicitTxContext;
-import org.apache.ignite.internal.sql.engine.framework.NoOpTransaction;
-import org.apache.ignite.internal.sql.engine.sql.IgniteSqlStartTransaction;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContext;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionContextImpl;
-import org.apache.ignite.internal.sql.engine.tx.QueryTransactionWrapper;
-import org.apache.ignite.internal.sql.engine.tx.ScriptTransactionContext;
-import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
-import org.apache.ignite.internal.tx.HybridTimestampTracker;
-import org.apache.ignite.internal.tx.TxManager;
-import org.apache.ignite.internal.tx.impl.TransactionInflights;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-/**
- * Tests for {@link RunningQueriesRegistry}.
- */
-@ExtendWith(MockitoExtension.class)
-public class RunningQueriesRegistryTest extends BaseIgniteAbstractTest {
- private final RunningQueriesRegistry registry = new RunningQueriesRegistryImpl();
-
- @Mock
- private TxManager txManager;
-
- @Mock
- private TransactionInflights transactionInflights;
-
- @Mock
- private HybridTimestampTracker observableTimeTracker;
-
- @Test
- public void registerQuery() {
- NoOpTransaction tx = NoOpTransaction.readWrite("RW");
- QueryTransactionContext txContext = ExplicitTxContext.fromTx(tx);
-
- RunningQueryInfo queryInfo = registry.registerQuery("test", "SELECT 1;", txContext.explicitTx());
-
- assertThat(registry.queries(), hasSize(1));
- assertSame(queryInfo, registry.queries().iterator().next());
-
- assertThat(queryInfo.queryId(), notNullValue());
- assertThat(queryInfo.phase(), equalTo(QueryExecutionPhase.INITIALIZATION.name()));
- assertThat(queryInfo.startTime(), lessThanOrEqualTo(Instant.now()));
- assertThat(queryInfo.queryType(), nullValue());
- assertThat(queryInfo.sql(), equalTo("SELECT 1;"));
- assertThat(queryInfo.schema(), equalTo("test"));
- assertThat(queryInfo.parentId(), nullValue());
- assertThat(queryInfo.statementNumber(), nullValue());
- assertThat(queryInfo.transactionId(), equalTo(tx.id().toString()));
-
- registry.unregister(queryInfo.queryId());
-
- assertThat(registry.queries(), hasSize(0));
- assertThat(registry.openedCursorsCount(), is(0));
- }
-
- @Test
- public void registerScriptWithTx() {
- String schema = "test";
-
- QueryTransactionContext txContext = new QueryTransactionContextImpl(txManager, observableTimeTracker, null, transactionInflights);
-
- when(txManager.begin(any(), anyBoolean())).thenReturn(NoOpTransaction.readWrite("node1"));
-
- ScriptTransactionContext scriptTxContext = new ScriptTransactionContext(txContext, transactionInflights);
-
- RunningScriptInfoTracker scriptInfoTracker = registry.registerScript(schema, "SELECT 1;", scriptTxContext);
-
- assertThat(registry.queries(), hasSize(1));
-
- RunningQueryInfo scriptQueryInfo = registry.queries().iterator().next();
-
- // Verify initial script query info.
- {
- assertThat(scriptQueryInfo.queryId(), notNullValue());
- assertThat(scriptQueryInfo.phase(), equalTo(QueryExecutionPhase.INITIALIZATION.name()));
- assertThat(scriptQueryInfo.startTime(), lessThanOrEqualTo(Instant.now()));
- assertThat(scriptQueryInfo.queryType(), equalTo(SCRIPT_QUERY_TYPE));
- assertThat(scriptQueryInfo.sql(), equalTo("SELECT 1;"));
- assertThat(scriptQueryInfo.schema(), equalTo(schema));
- assertThat(scriptQueryInfo.parentId(), nullValue());
- assertThat(scriptQueryInfo.statementNumber(), nullValue());
- assertThat(scriptQueryInfo.transactionId(), nullValue());
- }
-
- // After initialization, the script must be moved to the execution phase.
- scriptInfoTracker.onInitializationComplete(2);
- assertThat(scriptQueryInfo.phase(), equalTo(QueryExecutionPhase.EXECUTION.name()));
-
- // Simulate transaction start within script.
- scriptTxContext.handleControlStatement(Mockito.mock(IgniteSqlStartTransaction.class));
-
- QueryTransactionWrapper txWrapper = scriptTxContext.explicitTx();
- assertThat(txWrapper, notNullValue());
-
- String expectedTxId = txWrapper.unwrap().id().toString();
-
- RunningQueryInfo statement1 = scriptInfoTracker.registerStatement("SELECT 1", SqlQueryType.QUERY);
- assertThat(registry.queries(), hasSize(2));
-
- RunningQueryInfo statement2 = scriptInfoTracker.registerStatement("SELECT 2", SqlQueryType.QUERY);
- assertThat(registry.queries(), hasSize(3));
-
- // Verify script statements info.
- {
- assertThat(statement1.queryId(), notNullValue());
- assertThat(statement2.queryId(), notNullValue());
-
- assertThat(statement1.queryId(), not(equalTo(statement2.queryId())));
-
- assertThat(statement1.phase(), equalTo(QueryExecutionPhase.INITIALIZATION.name()));
- assertThat(statement2.phase(), equalTo(QueryExecutionPhase.INITIALIZATION.name()));
-
- assertThat(statement1.startTime(), lessThanOrEqualTo(Instant.now()));
- assertThat(statement2.startTime(), lessThanOrEqualTo(Instant.now()));
-
- assertThat(statement1.queryType(), equalTo(SqlQueryType.QUERY.name()));
- assertThat(statement2.queryType(), equalTo(SqlQueryType.QUERY.name()));
-
- assertThat(statement1.sql(), equalTo("SELECT 1"));
- assertThat(statement2.sql(), equalTo("SELECT 2"));
-
- assertThat(statement1.schema(), equalTo(schema));
- assertThat(statement2.schema(), equalTo(schema));
-
- assertThat(statement1.parentId(), equalTo(scriptQueryInfo.queryId().toString()));
- assertThat(statement2.parentId(), equalTo(scriptQueryInfo.queryId().toString()));
-
- assertThat(statement1.statementNumber(), is(1));
- assertThat(statement2.statementNumber(), is(2));
-
- assertThat(statement1.transactionId(), equalTo(expectedTxId));
- assertThat(statement2.transactionId(), equalTo(expectedTxId));
- }
-
- // Simulate cursor registration.
- AsyncSqlCursor<?> cursor = Mockito.mock(AsyncSqlCursor.class);
- CompletableFuture<Void> closeFut = new CompletableFuture<>();
- when(cursor.onClose()).thenReturn(closeFut);
-
- registry.registerCursor(statement2, cursor);
- assertThat(registry.openedCursorsCount(), is(1));
- assertThat(registry.queries(), hasSize(3));
-
- // Closing the cursor must unregister the statement.
- closeFut.complete(null);
- assertThat(registry.openedCursorsCount(), is(0));
- assertThat(registry.queries(), hasSize(2));
-
- // When the last script statement unregisters itself,
- // the script query must be unregistered as well.
- registry.unregister(statement1.queryId());
- assertThat(registry.queries(), hasSize(0));
- }
-
- @Test
- public void registerScriptWithExplicitTx() {
- String schema = "test";
-
- NoOpTransaction tx = NoOpTransaction.readWrite("node1");
- QueryTransactionContext txContext = new QueryTransactionContextImpl(txManager, observableTimeTracker, tx, transactionInflights);
-
- ScriptTransactionContext scriptTxContext = new ScriptTransactionContext(txContext, transactionInflights);
-
- RunningScriptInfoTracker scriptInfoTracker = registry.registerScript(schema, "SELECT 1;", scriptTxContext);
-
- assertThat(registry.queries(), hasSize(1));
-
- scriptInfoTracker.onInitializationComplete(2);
-
- RunningQueryInfo scriptInfo = registry.queries().iterator().next();
- RunningQueryInfo stmt1 = scriptInfoTracker.registerStatement("SELECT 1", SqlQueryType.QUERY);
- RunningQueryInfo stmt2 = scriptInfoTracker.registerStatement("SELECT 2", SqlQueryType.QUERY);
-
- assertThat(scriptInfo.transactionId(), equalTo(tx.id().toString()));
- assertThat(stmt1.transactionId(), equalTo(tx.id().toString()));
- assertThat(stmt2.transactionId(), equalTo(tx.id().toString()));
-
- registry.unregister(stmt1.queryId());
- registry.unregister(stmt2.queryId());
-
- assertThat(registry.queries(), hasSize(0));
- }
-}