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 e4a495634b IGNITE-17685 Sql. SQL engine should send messages
asynchronously (#1980)
e4a495634b is described below
commit e4a495634ba7865f4e8a982f24d1783f65850dbd
Author: korlov42 <[email protected]>
AuthorDate: Thu Apr 27 09:20:53 2023 +0300
IGNITE-17685 Sql. SQL engine should send messages asynchronously (#1980)
---
.../java/org/apache/ignite/lang/ErrorGroups.java | 6 +-
.../apache/ignite/internal/sql/engine/Stubs.java | 25 ---
.../internal/sql/engine/exec/ExchangeService.java | 98 +++++------
.../sql/engine/exec/ExchangeServiceImpl.java | 68 ++------
.../sql/engine/exec/ExecutionServiceImpl.java | 192 +++++++++++++--------
.../internal/sql/engine/exec/MailboxRegistry.java | 12 --
.../sql/engine/exec/MailboxRegistryImpl.java | 31 ----
.../ignite/internal/sql/engine/exec/rel/Inbox.java | 29 ++--
.../internal/sql/engine/exec/rel/Outbox.java | 65 +++++--
.../sql/engine/message/InboxCloseMessage.java | 45 -----
.../sql/engine/message/MessageService.java | 12 +-
.../sql/engine/message/MessageServiceImpl.java | 31 ++--
.../sql/engine/message/SqlQueryMessageGroup.java | 4 +-
.../ignite/internal/sql/engine/prepare/Cloner.java | 2 +-
.../internal/sql/engine/prepare/FragmentPlan.java | 65 -------
.../internal/sql/engine/StopCalciteModuleTest.java | 1 -
.../sql/engine/exec/ExecutionServiceImplTest.java | 58 +++----
.../internal/sql/engine/framework/TestNode.java | 1 -
18 files changed, 272 insertions(+), 473 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index 7ddb303ad3..2fd6edc2fb 100755
--- a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -187,9 +187,6 @@ public class ErrorGroups {
/** Node left the cluster. */
public static final int NODE_LEFT_ERR =
SQL_ERR_GROUP.registerErrorCode(27);
- /** Message send error. */
- public static final int MESSAGE_SEND_ERR =
SQL_ERR_GROUP.registerErrorCode(28);
-
/** Operation aborted/interrupted error. */
public static final int OPERATION_INTERRUPTED_ERR =
SQL_ERR_GROUP.registerErrorCode(29);
@@ -204,6 +201,9 @@ public class ErrorGroups {
/** Execution cancelled. */
public static final int EXECUTION_CANCELLED_ERR =
SQL_ERR_GROUP.registerErrorCode(33);
+
+ /** Error code describing any unexpected situation. */
+ public static final int INTERNAL_ERR =
SQL_ERR_GROUP.registerErrorCode(34);
}
/** Meta storage error group. */
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/Stubs.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/Stubs.java
deleted file mode 100644
index e83bb5785e..0000000000
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/Stubs.java
+++ /dev/null
@@ -1,25 +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;
-
-/** Stubs. */
-public class Stubs {
- public static int intFoo(Object... args) {
- return args == null ? 0 : args.length;
- }
-}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
index acd62415a0..e038afd632 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
@@ -19,77 +19,61 @@ package org.apache.ignite.internal.sql.engine.exec;
import java.util.List;
import java.util.UUID;
-import org.apache.ignite.lang.IgniteInternalCheckedException;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.sql.engine.exec.rel.Inbox;
+import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
import org.jetbrains.annotations.Nullable;
/**
- * ExchangeService interface.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * The interface describes a minimal but sufficient set of methods to make
+ * the exchange of information between mailboxes work.
+ *
+ * @see MailboxRegistry
+ * @see Outbox
+ * @see Inbox
*/
public interface ExchangeService extends LifecycleAware {
/**
- * Sends a batch of data to remote node.
- *
- * @param nodeName Target node consistent ID.
- * @param qryId Query ID.
- * @param fragmentId Target fragment ID.
- * @param exchangeId Exchange ID.
- * @param batchId Batch ID.
- * @param last Last batch flag.
- * @param rows Data rows.
- */
- <RowT> void sendBatch(String nodeName, UUID qryId, long fragmentId, long
exchangeId, int batchId, boolean last,
- List<RowT> rows) throws IgniteInternalCheckedException;
-
- /**
- * Requests batches from remote source.
- *
- * @param nodeName A consistent identifier of the node to request from.
- * @param queryId An identifier of the query.
- * @param fragmentId An identifier of the fragment to request from.
- * @param exchangeId An identifier of the exchange to request from.
- * @param amountOfBatches A count of batches to request.
- * @param state A state to propagate to the remote source.
- */
- void request(String nodeName, UUID queryId, long fragmentId, long
exchangeId, int amountOfBatches,
- @Nullable SharedState state) throws IgniteInternalCheckedException;
-
- /**
- * Sends cancel request.
- *
- * @param nodeName Target node consistent ID.
- * @param qryId Query ID.
- * @param fragmentId Target fragment ID.
- * @param exchangeId Exchange ID.
- */
- void closeInbox(String nodeName, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException;
-
- /**
- * Sends cancel request.
+ * Asynchronously sends a batch of data to the specified node.
*
- * @param nodeName Target node consistent ID.
- * @param qryId Query ID.
+ * @param nodeName The name of the node to which the data will be sent.
+ * @param queryId The ID of the query to which the data belongs.
+ * @param fragmentId The ID of the fragment to which the data will be sent.
+ * @param exchangeId The ID of the exchange through which the data will be
sent.
+ * @param batchId The ID of the batch to which the data belongs.
+ * @param last Indicates whether this is the last batch of data to be sent.
+ * @param rows The data to be sent.
+ * @param <RowT> The type of the rows int the batch.
+ * @return A {@link CompletableFuture future} representing the result of
operation,
+ * which completes when the data has been sent.
*/
- void closeQuery(String nodeName, UUID qryId) throws
IgniteInternalCheckedException;
+ <RowT> CompletableFuture<Void> sendBatch(String nodeName, UUID queryId,
long fragmentId, long exchangeId, int batchId, boolean last,
+ List<RowT> rows);
/**
- * Send error.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Asynchronously requests data from the specified node.
*
- * @param nodeName Target node consistent ID.
- * @param qryId Query ID.
- * @param fragmentId Source fragment ID.
- * @param err Exception to send.
- * @throws IgniteInternalCheckedException On error marshaling or send
ErrorMessage.
+ * @param nodeName The name of the node from which the data will be
requested.
+ * @param queryId The ID of the query for which the data is being
requested.
+ * @param fragmentId The ID of the fragment from which the data will be
requested.
+ * @param exchangeId The ID of the exchange through which the data will be
requested.
+ * @param amountOfBatches The number of batches of data to request.
+ * @param state The state to propagate to the remote node, or null if
state is not changed or not required.
+ * @return A {@link CompletableFuture future} representing the result of
operation,
+ * which completes when the request message has been sent.
*/
- void sendError(String nodeName, UUID qryId, long fragmentId, Throwable
err) throws IgniteInternalCheckedException;
+ CompletableFuture<Void> request(String nodeName, UUID queryId, long
fragmentId, long exchangeId, int amountOfBatches,
+ @Nullable SharedState state);
/**
- * Alive.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
+ * Asynchronously sends an error message to the specified node.
*
- * @param nodeName Target node consistent ID.
- * @return {@code true} if node is alive, {@code false} otherwise.
+ * @param nodeName The name of the node to which the error will be sent.
+ * @param queryId The ID of the query to which the error belongs.
+ * @param fragmentId The ID of the fragment to which the error belongs.
+ * @param error The error to send.
+ * @return A {@link CompletableFuture future} representing the result of
operation,
+ * which completes when the error message has been sent.
*/
- boolean alive(String nodeName);
+ CompletableFuture<Void> sendError(String nodeName, UUID queryId, long
fragmentId, Throwable error);
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
index e83a2b2ae1..a25cc6bfe9 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
@@ -17,10 +17,8 @@
package org.apache.ignite.internal.sql.engine.exec;
-import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.lang.ErrorGroups.Common.UNEXPECTED_ERR;
-import java.util.Collection;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -29,14 +27,12 @@ import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.exec.rel.Inbox;
import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
-import org.apache.ignite.internal.sql.engine.message.InboxCloseMessage;
import org.apache.ignite.internal.sql.engine.message.MessageService;
import org.apache.ignite.internal.sql.engine.message.QueryBatchMessage;
import org.apache.ignite.internal.sql.engine.message.QueryBatchRequestMessage;
import org.apache.ignite.internal.sql.engine.message.SqlQueryMessageGroup;
import org.apache.ignite.internal.sql.engine.message.SqlQueryMessagesFactory;
import org.apache.ignite.internal.sql.engine.util.Commons;
-import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
import org.jetbrains.annotations.Nullable;
@@ -69,16 +65,15 @@ public class ExchangeServiceImpl implements ExchangeService
{
/** {@inheritDoc} */
@Override
public void start() {
- messageService.register((n, m) -> onMessage(n, (InboxCloseMessage) m),
SqlQueryMessageGroup.INBOX_CLOSE_MESSAGE);
messageService.register((n, m) -> onMessage(n,
(QueryBatchRequestMessage) m), SqlQueryMessageGroup.QUERY_BATCH_REQUEST);
messageService.register((n, m) -> onMessage(n, (QueryBatchMessage) m),
SqlQueryMessageGroup.QUERY_BATCH_MESSAGE);
}
/** {@inheritDoc} */
@Override
- public <RowT> void sendBatch(String nodeName, UUID qryId, long fragmentId,
long exchangeId, int batchId,
- boolean last, List<RowT> rows) throws
IgniteInternalCheckedException {
- messageService.send(
+ public <RowT> CompletableFuture<Void> sendBatch(String nodeName, UUID
qryId, long fragmentId, long exchangeId, int batchId,
+ boolean last, List<RowT> rows) {
+ return messageService.send(
nodeName,
FACTORY.queryBatchMessage()
.queryId(qryId)
@@ -93,9 +88,9 @@ public class ExchangeServiceImpl implements ExchangeService {
/** {@inheritDoc} */
@Override
- public void request(String nodeName, UUID queryId, long fragmentId, long
exchangeId, int amountOfBatches,
- @Nullable SharedState state) throws IgniteInternalCheckedException
{
- messageService.send(
+ public CompletableFuture<Void> request(String nodeName, UUID queryId, long
fragmentId, long exchangeId, int amountOfBatches,
+ @Nullable SharedState state) {
+ return messageService.send(
nodeName,
FACTORY.queryBatchRequestMessage()
.queryId(queryId)
@@ -109,60 +104,17 @@ public class ExchangeServiceImpl implements
ExchangeService {
/** {@inheritDoc} */
@Override
- public void closeQuery(String nodeName, UUID qryId) throws
IgniteInternalCheckedException {
- messageService.send(
- nodeName,
- FACTORY.queryCloseMessage()
- .queryId(qryId)
- .build()
- );
- }
-
- /** {@inheritDoc} */
- @Override
- public void closeInbox(String nodeName, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException {
- messageService.send(
- nodeName,
- FACTORY.inboxCloseMessage()
- .queryId(qryId)
- .fragmentId(fragmentId)
- .exchangeId(exchangeId)
- .build()
- );
- }
-
- /** {@inheritDoc} */
- @Override
- public void sendError(String nodeName, UUID qryId, long fragmentId,
Throwable err) throws IgniteInternalCheckedException {
- messageService.send(
+ public CompletableFuture<Void> sendError(String nodeName, UUID queryId,
long fragmentId, Throwable error) {
+ return messageService.send(
nodeName,
FACTORY.errorMessage()
- .queryId(qryId)
+ .queryId(queryId)
.fragmentId(fragmentId)
- .error(err)
+ .error(error)
.build()
);
}
- /** {@inheritDoc} */
- @Override
- public boolean alive(String nodeName) {
- return messageService.alive(nodeName);
- }
-
- private void onMessage(String nodeName, InboxCloseMessage msg) {
- Collection<Inbox<?>> inboxes = mailboxRegistry.inboxes(msg.queryId(),
msg.fragmentId(), msg.exchangeId());
-
- if (!nullOrEmpty(inboxes)) {
- for (Inbox<?> inbox : inboxes) {
- inbox.context().execute(inbox::close, inbox::onError);
- }
- } else if (LOG.isDebugEnabled()) {
- LOG.debug("Stale inbox cancel message received [nodeName={},
queryId={}, fragmentId={}, exchangeId={}]",
- nodeName, msg.queryId(), msg.fragmentId(),
msg.exchangeId());
- }
- }
-
private void onMessage(String nodeName, QueryBatchRequestMessage msg) {
CompletableFuture<Outbox<?>> outboxFut =
mailboxRegistry.outbox(msg.queryId(), msg.exchangeId());
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 4df81cce82..76c2404894 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
@@ -21,8 +21,8 @@ import static
org.apache.ignite.internal.sql.engine.externalize.RelJsonReader.fr
import static
org.apache.ignite.internal.sql.engine.util.Commons.FRAMEWORK_CONFIG;
import static org.apache.ignite.internal.util.CollectionUtils.nullOrEmpty;
import static org.apache.ignite.lang.ErrorGroups.Sql.DDL_EXEC_ERR;
-import static org.apache.ignite.lang.ErrorGroups.Sql.MESSAGE_SEND_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.NODE_LEFT_ERR;
+import static org.apache.ignite.lang.IgniteStringFormatter.format;
import com.github.benmanes.caffeine.cache.Caffeine;
import java.util.ArrayList;
@@ -63,10 +63,10 @@ import
org.apache.ignite.internal.sql.engine.metadata.MappingService;
import org.apache.ignite.internal.sql.engine.metadata.MappingServiceImpl;
import org.apache.ignite.internal.sql.engine.metadata.NodeWithTerm;
import org.apache.ignite.internal.sql.engine.metadata.RemoteException;
+import org.apache.ignite.internal.sql.engine.prepare.Cloner;
import org.apache.ignite.internal.sql.engine.prepare.DdlPlan;
import org.apache.ignite.internal.sql.engine.prepare.ExplainPlan;
import org.apache.ignite.internal.sql.engine.prepare.Fragment;
-import org.apache.ignite.internal.sql.engine.prepare.FragmentPlan;
import org.apache.ignite.internal.sql.engine.prepare.IgniteRelShuttle;
import org.apache.ignite.internal.sql.engine.prepare.MappingQueryContext;
import org.apache.ignite.internal.sql.engine.prepare.MultiStepPlan;
@@ -84,6 +84,8 @@ import
org.apache.ignite.internal.sql.engine.util.HashFunctionFactoryImpl;
import org.apache.ignite.internal.sql.engine.util.TypeUtils;
import
org.apache.ignite.internal.table.distributed.replicator.TablePartitionId;
import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
@@ -108,7 +110,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
private static final SqlQueryMessagesFactory FACTORY = new
SqlQueryMessagesFactory();
- private final MessageService msgSrvc;
+ private final MessageService messageService;
private final TopologyService topSrvc;
@@ -120,8 +122,6 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
private final MappingService mappingSrvc;
- private final ExchangeService exchangeSrvc;
-
private final RowHandler<RowT> handler;
private final DdlCommandHandler ddlCmdHnd;
@@ -162,7 +162,6 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
ddlCommandHandler,
taskExecutor,
handler,
- exchangeSrvc,
ctx -> new LogicalRelImplementor<>(
ctx,
new HashFunctionFactoryImpl<>(sqlSchemaManager,
handler),
@@ -175,35 +174,32 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
/**
* Constructor.
*
- * @param msgSrvc Message service.
+ * @param messageService Message service.
* @param topSrvc Topology service.
* @param mappingSrvc Nodes mapping calculation service.
* @param sqlSchemaManager Schema manager.
* @param ddlCmdHnd Handler of the DDL commands.
* @param taskExecutor Task executor.
* @param handler Row handler.
- * @param exchangeSrvc Exchange service.
* @param implementorFactory Relational node implementor factory.
*/
public ExecutionServiceImpl(
- MessageService msgSrvc,
+ MessageService messageService,
TopologyService topSrvc,
MappingService mappingSrvc,
SqlSchemaManager sqlSchemaManager,
DdlCommandHandler ddlCmdHnd,
QueryTaskExecutor taskExecutor,
RowHandler<RowT> handler,
- ExchangeService exchangeSrvc,
ImplementorFactory<RowT> implementorFactory
) {
this.localNode = topSrvc.localMember();
this.handler = handler;
- this.msgSrvc = msgSrvc;
+ this.messageService = messageService;
this.mappingSrvc = mappingSrvc;
this.topSrvc = topSrvc;
this.sqlSchemaManager = sqlSchemaManager;
this.taskExecutor = taskExecutor;
- this.exchangeSrvc = exchangeSrvc;
this.ddlCmdHnd = ddlCmdHnd;
this.implementorFactory = implementorFactory;
}
@@ -211,10 +207,10 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
/** {@inheritDoc} */
@Override
public void start() {
- msgSrvc.register((n, m) -> onMessage(n, (QueryStartRequest) m),
SqlQueryMessageGroup.QUERY_START_REQUEST);
- msgSrvc.register((n, m) -> onMessage(n, (QueryStartResponse) m),
SqlQueryMessageGroup.QUERY_START_RESPONSE);
- msgSrvc.register((n, m) -> onMessage(n, (QueryCloseMessage) m),
SqlQueryMessageGroup.QUERY_CLOSE_MESSAGE);
- msgSrvc.register((n, m) -> onMessage(n, (ErrorMessage) m),
SqlQueryMessageGroup.ERROR_MESSAGE);
+ messageService.register((n, m) -> onMessage(n, (QueryStartRequest) m),
SqlQueryMessageGroup.QUERY_START_REQUEST);
+ messageService.register((n, m) -> onMessage(n, (QueryStartResponse)
m), SqlQueryMessageGroup.QUERY_START_RESPONSE);
+ messageService.register((n, m) -> onMessage(n, (QueryCloseMessage) m),
SqlQueryMessageGroup.QUERY_CLOSE_MESSAGE);
+ messageService.register((n, m) -> onMessage(n, (ErrorMessage) m),
SqlQueryMessageGroup.ERROR_MESSAGE);
}
private AsyncCursor<List<Object>> executeQuery(
@@ -246,10 +242,10 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
.build();
}
- private QueryPlan prepareFragment(String jsonFragment, BaseQueryContext
ctx) {
+ private IgniteRel relationalTreeFromJsonString(String jsonFragment,
BaseQueryContext ctx) {
IgniteRel plan = physNodesCache.computeIfAbsent(jsonFragment, ser ->
fromJson(ctx, ser));
- return new FragmentPlan(plan);
+ return new Cloner(Commons.cluster()).visit(plan);
}
/** {@inheritDoc} */
@@ -383,12 +379,6 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
f.join();
}
- /** {@inheritDoc} */
- @Override
- public void onAppeared(ClusterNode member) {
- // NO_OP
- }
-
/** {@inheritDoc} */
@Override
public void onDisappeared(ClusterNode member) {
@@ -419,19 +409,17 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
}
private DistributedQueryManager getOrCreateQueryManager(QueryStartRequest
msg) {
- DistributedQueryManager queryManager =
queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
+ return queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
BaseQueryContext ctx = createQueryContext(key, msg.schema(),
msg.parameters());
return new DistributedQueryManager(ctx);
});
-
- return queryManager;
}
/**
* A convenient class that manages the initialization and termination of
distributed queries.
*/
- class DistributedQueryManager {
+ private class DistributedQueryManager {
private final BaseQueryContext ctx;
private final CompletableFuture<Void> cancelFut = new
CompletableFuture<>();
@@ -465,10 +453,10 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
return List.copyOf(localFragments);
}
- private void sendFragment(
+ private CompletableFuture<Void> sendFragment(
String targetNodeName, Fragment fragment, FragmentDescription
desc, TxAttributes txAttributes
- ) throws IgniteInternalCheckedException {
- QueryStartRequest req = FACTORY.queryStartRequest()
+ ) {
+ QueryStartRequest request = FACTORY.queryStartRequest()
.queryId(ctx.queryId())
.fragmentId(fragment.fragmentId())
.schema(ctx.schemaName())
@@ -479,16 +467,14 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
.schemaVersion(ctx.schemaVersion())
.build();
- var fut = new CompletableFuture<Void>();
- remoteFragmentInitCompletion.put(new
RemoteFragmentKey(targetNodeName, fragment.fragmentId()), fut);
+ CompletableFuture<Void>
remoteFragmentInitializationCompletionFuture = new CompletableFuture<>();
- try {
- msgSrvc.send(targetNodeName, req);
- } catch (Exception ex) {
- fut.complete(null);
+ remoteFragmentInitCompletion.put(
+ new RemoteFragmentKey(targetNodeName,
fragment.fragmentId()),
+ remoteFragmentInitializationCompletionFuture
+ );
- throw ex;
- }
+ return messageService.send(targetNodeName, request);
}
private void acknowledgeFragment(String nodeName, long fragmentId,
@Nullable Throwable ex) {
@@ -525,15 +511,15 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
NODE_LEFT_ERR, "Node left the cluster
[nodeName=" + nodeName + "]")));
}
- private void executeFragment(FragmentPlan plan, ExecutionContext<RowT>
ectx) {
+ private CompletableFuture<Void> executeFragment(IgniteRel treeRoot,
ExecutionContext<RowT> ectx) {
String origNodeName = ectx.originatingNodeName();
- AbstractNode<RowT> node =
implementorFactory.create(ectx).go(plan.root());
+ AbstractNode<RowT> node =
implementorFactory.create(ectx).go(treeRoot);
localFragments.add(node);
if (!(node instanceof Outbox)) {
- Function<RowT, RowT> internalTypeConverter =
TypeUtils.resultTypeConverter(ectx, plan.root().getRowType());
+ Function<RowT, RowT> internalTypeConverter =
TypeUtils.resultTypeConverter(ectx, treeRoot.getRowType());
AsyncRootNode<RowT, List<Object>> rootNode = new
AsyncRootNode<>(node, inRow -> {
inRow = internalTypeConverter.apply(inRow);
@@ -555,21 +541,19 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
root.complete(rootNode);
}
- try {
- msgSrvc.send(
- origNodeName,
- FACTORY.queryStartResponse()
- .queryId(ectx.queryId())
- .fragmentId(ectx.fragmentId())
- .build()
- );
- } catch (IgniteInternalCheckedException e) {
- throw new IgniteInternalException(MESSAGE_SEND_ERR, "Failed to
send reply. [nodeName=" + origNodeName + ']', e);
- }
+ CompletableFuture<Void> sendingResult = messageService.send(
+ origNodeName,
+ FACTORY.queryStartResponse()
+ .queryId(ectx.queryId())
+ .fragmentId(ectx.fragmentId())
+ .build()
+ );
if (node instanceof Outbox) {
((Outbox<?>) node).prefetch();
}
+
+ return sendingResult;
}
private ExecutionContext<RowT> createContext(String initiatorNodeName,
FragmentDescription desc, TxAttributes txAttributes) {
@@ -592,22 +576,26 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
FragmentDescription desc,
TxAttributes txAttributes
) {
- try {
- QueryPlan qryPlan = prepareFragment(fragmentString, ctx);
+ CompletableFuture<?> start = new CompletableFuture<>();
- FragmentPlan plan = (FragmentPlan) qryPlan;
+ start.thenCompose(none -> {
+ IgniteRel treeRoot =
relationalTreeFromJsonString(fragmentString, ctx);
- executeFragment(plan, createContext(initiatorNode, desc,
txAttributes));
- } catch (Throwable ex) {
+ return executeFragment(treeRoot, createContext(initiatorNode,
desc, txAttributes));
+ }).exceptionally(ex -> {
handleError(ex, initiatorNode, desc.fragmentId());
- }
+
+ return null;
+ });
+
+ start.complete(null);
}
private void handleError(Throwable ex, String initiatorNode, long
fragmentId) {
LOG.debug("Unable to start query fragment", ex);
try {
- msgSrvc.send(
+ messageService.send(
initiatorNode,
FACTORY.queryStartResponse()
.queryId(ctx.queryId())
@@ -644,6 +632,8 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
// of the commit partition (if there is at least one table
in the fragments)
TxAttributes attributes = TxAttributes.fromTx(tx);
+ List<CompletableFuture<?>> resultsOfFragmentSending = new
ArrayList<>();
+
// start remote execution
for (Fragment fragment : fragments) {
if (fragment.rootFragment()) {
@@ -661,10 +651,68 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
);
for (String nodeName : fragmentDesc.nodeNames()) {
- sendFragment(nodeName, fragment, fragmentDesc,
attributes);
+ CompletableFuture<Void> resultOfSending =
sendFragment(nodeName, fragment, fragmentDesc, attributes);
+
+ resultsOfFragmentSending.add(resultOfSending);
+
+ resultOfSending.handle((ignored, t) -> {
+ if (t == null) {
+ return null;
+ }
+
+ // if we were unable to send a request, then
no need
+ // to wait for the remote node to complete
initialization
+
+ CompletableFuture<?> completionFuture =
remoteFragmentInitCompletion.get(
+ new RemoteFragmentKey(nodeName,
fragment.fragmentId())
+ );
+
+ if (completionFuture != null) {
+ completionFuture.complete(null);
+ }
+
+ throw ExceptionUtils.withCauseAndCode(
+ IgniteInternalException::new,
+ Sql.INTERNAL_ERR,
+ format("Unable to send fragment
[targetNode={}, fragmentId={}, cause={}]",
+ nodeName,
fragment.fragmentId(), t.getMessage()),
+ t
+ );
+ });
}
}
+
+
CompletableFuture.allOf(resultsOfFragmentSending.toArray(new
CompletableFuture[0]))
+ .handle((ignoredVal, ignoredTh) -> {
+ if (ignoredTh == null) {
+ return null;
+ }
+
+ Throwable firstFoundError = null;
+
+ for (CompletableFuture<?> fut :
resultsOfFragmentSending) {
+ if (fut.isCompletedExceptionally()) {
+ // this is non blocking join() because
we are inside of CompletableFuture.allOf call
+ Throwable fromFuture =
fut.handle((ignored, ex) -> ex).join();
+
+ if (firstFoundError == null) {
+ firstFoundError = fromFuture;
+ } else {
+
firstFoundError.addSuppressed(fromFuture);
+ }
+ }
+ }
+
+ Throwable error = firstFoundError;
+ if (!root.completeExceptionally(error)) {
+ root.thenAccept(root ->
root.onError(error));
+ }
+
+ return null;
+ });
} catch (Throwable t) {
+ LOG.warn("Unexpected exception during query
initialization", t);
+
if (!root.completeExceptionally(t)) {
root.thenAccept(root -> root.onError(t));
}
@@ -766,24 +814,22 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
for (Map.Entry<String, List<CompletableFuture<?>>>
entry : requestsPerNode.entrySet()) {
String nodeId = entry.getKey();
- if (!exchangeSrvc.alive(nodeId)) {
- continue;
- }
-
cancelFuts.add(
CompletableFuture.allOf(entry.getValue().toArray(new CompletableFuture[0]))
.handle((none2, t) -> {
- // t is ignored in this block
because it's passed to a cursor.
-
- try {
-
exchangeSrvc.closeQuery(nodeId, ctx.queryId());
- } catch
(IgniteInternalCheckedException e) {
- throw new
IgniteInternalException(MESSAGE_SEND_ERR,
- "Failed to send
cancel message. [nodeId=" + nodeId + ']', e);
- }
+ // Some fragments may be
completed exceptionally, and that is fine.
+ // Here we need to make sure
that all query initialisation requests
+ // have been processed before
sending a cancellation request. That's
+ // why we just ignore any
exception.
return null;
})
+ .thenCompose(ignored ->
messageService.send(
+ nodeId,
+ FACTORY.queryCloseMessage()
+
.queryId(ctx.queryId())
+ .build()
+ ))
);
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistry.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistry.java
index eed6a5d0ba..75676d435f 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistry.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistry.java
@@ -17,12 +17,10 @@
package org.apache.ignite.internal.sql.engine.exec;
-import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.sql.engine.exec.rel.Inbox;
import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
-import org.jetbrains.annotations.Nullable;
/**
* MailboxRegistry interface.
@@ -74,14 +72,4 @@ public interface MailboxRegistry extends LifecycleAware {
* @return Registered inbox. May be {@code null} if execution was
cancelled.
*/
Inbox<?> inbox(UUID qryId, long exchangeId);
-
- /**
- * Returns all registered inboxes for provided query ID.
- *
- * @param qryId Query ID. {@code null} means return inboxes with any
query id.
- * @param fragmentId Fragment Id. {@code -1} means return inboxes with any
fragment id.
- * @param exchangeId Exchange Id. {@code -1} means return inboxes with any
exchange id.
- * @return Registered inboxes.
- */
- Collection<Inbox<?>> inboxes(@Nullable UUID qryId, long fragmentId, long
exchangeId);
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
index aa939b9769..5124f72a2a 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
@@ -17,23 +17,17 @@
package org.apache.ignite.internal.sql.engine.exec;
-import java.util.Collection;
import java.util.Map;
-import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.exec.rel.Inbox;
-import org.apache.ignite.internal.sql.engine.exec.rel.Mailbox;
import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.TopologyEventHandler;
-import org.jetbrains.annotations.Nullable;
/**
* MailboxRegistryImpl.
@@ -42,8 +36,6 @@ import org.jetbrains.annotations.Nullable;
public class MailboxRegistryImpl implements MailboxRegistry,
TopologyEventHandler {
private static final IgniteLogger LOG =
Loggers.forClass(MailboxRegistryImpl.class);
- private static final Predicate<Mailbox<?>> ALWAYS_TRUE = o -> true;
-
private final Map<MailboxKey, CompletableFuture<Outbox<?>>> locals;
private final Map<MailboxKey, Inbox<?>> remotes;
@@ -124,29 +116,6 @@ public class MailboxRegistryImpl implements
MailboxRegistry, TopologyEventHandle
return remotes.get(new MailboxKey(qryId, exchangeId));
}
- /** {@inheritDoc} */
- @Override
- public Collection<Inbox<?>> inboxes(@Nullable UUID qryId, long fragmentId,
long exchangeId) {
- return remotes.values().stream()
- .filter(makeFilter(qryId, fragmentId, exchangeId))
- .collect(Collectors.toList());
- }
-
- private static Predicate<Mailbox<?>> makeFilter(@Nullable UUID qryId, long
fragmentId, long exchangeId) {
- Predicate<Mailbox<?>> filter = ALWAYS_TRUE;
- if (qryId != null) {
- filter = filter.and(mailbox -> Objects.equals(mailbox.queryId(),
qryId));
- }
- if (fragmentId != -1) {
- filter = filter.and(mailbox -> mailbox.fragmentId() == fragmentId);
- }
- if (exchangeId != -1) {
- filter = filter.and(mailbox -> mailbox.exchangeId() == exchangeId);
- }
-
- return filter;
- }
-
/** {@inheritDoc} */
@Override
public String toString() {
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
index 2c7f63a2b7..81aade9a57 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
@@ -35,7 +35,10 @@ import
org.apache.ignite.internal.sql.engine.exec.ExecutionContext;
import org.apache.ignite.internal.sql.engine.exec.MailboxRegistry;
import org.apache.ignite.internal.sql.engine.exec.SharedState;
import org.apache.ignite.internal.sql.engine.exec.rel.Inbox.RemoteSource.State;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.lang.IgniteInternalCheckedException;
+import org.apache.ignite.lang.IgniteInternalException;
import org.jetbrains.annotations.Nullable;
/**
@@ -174,10 +177,6 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
private void push() throws Exception {
if (remoteSources == null) {
- for (String node : srcNodeNames) {
- checkNode(node);
- }
-
remoteSources = new ArrayList<>(srcNodeNames.size());
for (String nodeName : srcNodeNames) {
@@ -334,8 +333,20 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
}
}
- private void requestBatches(String nodeName, int cnt, @Nullable
SharedState state) throws IgniteInternalCheckedException {
- exchange.request(nodeName, queryId(), srcFragmentId, exchangeId, cnt,
state);
+ private void requestBatches(String nodeName, int cnt, @Nullable
SharedState state) {
+ exchange.request(nodeName, queryId(), srcFragmentId, exchangeId, cnt,
state)
+ .whenComplete((ignored, ex) -> {
+ if (ex != null) {
+ IgniteInternalException wrapperEx =
ExceptionUtils.withCauseAndCode(
+ IgniteInternalException::new,
+ Sql.INTERNAL_ERR,
+ "Unable to request next batch: " +
ex.getMessage(),
+ ex
+ );
+
+ context().execute(() -> onError(wrapperEx),
this::onError);
+ }
+ });
}
/**
@@ -358,12 +369,6 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
}
}
- private void checkNode(String nodeName) throws
IgniteInternalCheckedException {
- if (!exchange.alive(nodeName)) {
- throw new IgniteInternalCheckedException(NODE_LEFT_ERR, "Failed to
execute query, node left [nodeName=" + nodeName + ']');
- }
- }
-
private static final class Batch<RowT> implements Comparable<Batch<RowT>> {
private final int batchId;
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
index dad5cbb4f9..0e30d48d4e 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
@@ -26,6 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.UUID;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.sql.engine.exec.ExchangeService;
@@ -34,7 +35,10 @@ import
org.apache.ignite.internal.sql.engine.exec.MailboxRegistry;
import org.apache.ignite.internal.sql.engine.exec.SharedState;
import org.apache.ignite.internal.sql.engine.trait.Destination;
import org.apache.ignite.internal.sql.engine.util.Commons;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.lang.ErrorGroups.Sql;
import org.apache.ignite.lang.IgniteInternalCheckedException;
+import org.apache.ignite.lang.IgniteInternalException;
import org.jetbrains.annotations.Nullable;
/**
@@ -172,13 +176,9 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
/** {@inheritDoc} */
@Override
public void onError(Throwable e) {
- try {
- sendError(e);
- } catch (IgniteInternalCheckedException ex) {
- LOG.info("Unable to send error message", e);
- } finally {
- Commons.closeQuiet(this);
- }
+ sendError(e);
+
+ Commons.closeQuiet(this);
}
/** {@inheritDoc} */
@@ -231,20 +231,47 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
return this;
}
- private void sendBatch(String nodeName, int batchId, boolean last,
List<RowT> rows) throws IgniteInternalCheckedException {
- exchange.sendBatch(nodeName, queryId(), targetFragmentId, exchangeId,
batchId, last, rows);
+ private void sendBatch(String nodeName, int batchId, boolean last,
List<RowT> rows) {
+ exchange.sendBatch(nodeName, queryId(), targetFragmentId, exchangeId,
batchId, last, rows)
+ .whenComplete((ignored, ex) -> {
+ if (ex == null) {
+ return;
+ }
+
+ IgniteInternalException wrapperEx =
ExceptionUtils.withCauseAndCode(
+ IgniteInternalException::new,
+ Sql.INTERNAL_ERR,
+ "Unable to send batch: " + ex.getMessage(),
+ ex
+ );
+
+ context().execute(() -> onError(wrapperEx), this::onError);
+ });
}
- private void sendError(Throwable err) throws
IgniteInternalCheckedException {
- exchange.sendError(context().originatingNodeName(), queryId(),
fragmentId(), err);
- }
-
- private void sendInboxClose(String nodeName) {
- try {
- exchange.closeInbox(nodeName, queryId(), targetFragmentId,
exchangeId);
- } catch (IgniteInternalCheckedException e) {
- LOG.info("Unable to send cancel message", e);
- }
+ private void sendError(Throwable original) {
+ String nodeName = context().originatingNodeName();
+ UUID queryId = queryId();
+ long fragmentId = fragmentId();
+
+ exchange.sendError(nodeName, queryId, fragmentId, original)
+ .whenComplete((ignored, ex) -> {
+ if (ex == null) {
+ return;
+ }
+
+ IgniteInternalException wrapperEx =
ExceptionUtils.withCauseAndCode(
+ IgniteInternalException::new,
+ Sql.INTERNAL_ERR,
+ "Unable to send error: " + ex.getMessage(),
+ ex
+ );
+
+ wrapperEx.addSuppressed(original);
+
+ LOG.warn("Unable to send error to a remote node
[queryId={}, fragmentId={}, targetNode={}]",
+ queryId, fragmentId, nodeName, wrapperEx);
+ });
}
private void flush() throws Exception {
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/InboxCloseMessage.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/InboxCloseMessage.java
deleted file mode 100644
index b8c7364c3a..0000000000
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/InboxCloseMessage.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.message;
-
-import java.io.Serializable;
-import java.util.UUID;
-import org.apache.ignite.network.NetworkMessage;
-import org.apache.ignite.network.annotations.Transferable;
-
-/**
- * InboxCloseMessage interface.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
-@Transferable(value = SqlQueryMessageGroup.INBOX_CLOSE_MESSAGE)
-public interface InboxCloseMessage extends NetworkMessage, Serializable {
- /**
- * Get query ID.
- */
- UUID queryId();
-
- /**
- * Get fragment ID.
- */
- long fragmentId();
-
- /**
- * Get exchange ID.
- */
- long exchangeId();
-}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
index 633ef7f86d..c7bad33de1 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
@@ -17,8 +17,8 @@
package org.apache.ignite.internal.sql.engine.message;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.sql.engine.exec.LifecycleAware;
-import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.network.NetworkMessage;
/**
@@ -32,15 +32,7 @@ public interface MessageService extends LifecycleAware {
* @param nodeName Node consistent ID.
* @param msg Message.
*/
- void send(String nodeName, NetworkMessage msg) throws
IgniteInternalCheckedException;
-
- /**
- * Checks whether a node with given ID is alive.
- *
- * @param nodeName Node consistent ID.
- * @return {@code True} if node is alive.
- */
- boolean alive(String nodeName);
+ CompletableFuture<Void> send(String nodeName, NetworkMessage msg);
/**
* Registers a listener for messages of a given type.
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
index a0010e81b9..b144fa69ef 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
@@ -18,15 +18,14 @@
package org.apache.ignite.internal.sql.engine.message;
import static
org.apache.ignite.internal.sql.engine.message.SqlQueryMessageGroup.GROUP_TYPE;
-import static org.apache.ignite.lang.ErrorGroups.Sql.MESSAGE_SEND_ERR;
import static org.apache.ignite.lang.ErrorGroups.Sql.NODE_LEFT_ERR;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.sql.engine.exec.QueryTaskExecutor;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
-import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.MessagingService;
@@ -77,33 +76,29 @@ public class MessageServiceImpl implements MessageService {
/** {@inheritDoc} */
@Override
- public void send(String nodeName, NetworkMessage msg) throws
IgniteInternalCheckedException {
+ public CompletableFuture<Void> send(String nodeName, NetworkMessage msg) {
if (!busyLock.enterBusy()) {
- return;
+ return CompletableFuture.completedFuture(null);
}
try {
if (locNodeName.equals(nodeName)) {
onMessage(nodeName, msg);
+
+ return CompletableFuture.completedFuture(null);
} else {
ClusterNode node = topSrvc.getByConsistentId(nodeName);
if (node == null) {
- throw new IgniteInternalException(
+ return CompletableFuture.failedFuture(new
IgniteInternalException(
NODE_LEFT_ERR, "Failed to send message to node
(has node left grid?): " + nodeName
- );
+ ));
}
- try {
- messagingSrvc.send(node, msg).join();
- } catch (Exception ex) {
- if (ex instanceof IgniteInternalCheckedException) {
- throw (IgniteInternalCheckedException) ex;
- }
-
- throw new IgniteInternalCheckedException(MESSAGE_SEND_ERR,
ex);
- }
+ return messagingSrvc.send(node, msg);
}
+ } catch (Exception ex) {
+ return CompletableFuture.failedFuture(ex);
} finally {
busyLock.leaveBusy();
}
@@ -121,12 +116,6 @@ public class MessageServiceImpl implements MessageService {
assert old == null : old;
}
- /** {@inheritDoc} */
- @Override
- public boolean alive(String nodeName) {
- return topSrvc.getByConsistentId(nodeName) != null;
- }
-
private void onMessage(String consistentId, NetworkMessage msg) {
if (msg instanceof ExecutionContextAwareMessage) {
ExecutionContextAwareMessage msg0 = (ExecutionContextAwareMessage)
msg;
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/SqlQueryMessageGroup.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/SqlQueryMessageGroup.java
index d4cc366191..e8d8355f71 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/SqlQueryMessageGroup.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/SqlQueryMessageGroup.java
@@ -39,7 +39,5 @@ public final class SqlQueryMessageGroup {
/** See {@link QueryBatchRequestMessage} for details. */
public static final short QUERY_BATCH_REQUEST = 4;
- public static final short INBOX_CLOSE_MESSAGE = 5;
-
- public static final short QUERY_CLOSE_MESSAGE = 6;
+ public static final short QUERY_CLOSE_MESSAGE = 5;
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Cloner.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Cloner.java
index e6adf7a2f5..8433032cbc 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Cloner.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Cloner.java
@@ -33,7 +33,7 @@ public class Cloner {
private List<IgniteReceiver> remotes;
- Cloner(RelOptCluster cluster) {
+ public Cloner(RelOptCluster cluster) {
this.cluster = cluster;
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/FragmentPlan.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/FragmentPlan.java
deleted file mode 100644
index 658b0874b8..0000000000
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/FragmentPlan.java
+++ /dev/null
@@ -1,65 +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.prepare;
-
-import org.apache.calcite.plan.RelOptCluster;
-import org.apache.ignite.internal.sql.engine.SqlQueryType;
-import org.apache.ignite.internal.sql.engine.rel.IgniteRel;
-import org.apache.ignite.internal.sql.engine.util.Commons;
-import org.apache.ignite.sql.ResultSetMetadata;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * FragmentPlan.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
-public class FragmentPlan implements QueryPlan {
- private final IgniteRel root;
-
- /**
- * Constructor.
- * TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
- */
- public FragmentPlan(IgniteRel root) {
- RelOptCluster cluster = Commons.cluster();
-
- this.root = new Cloner(cluster).visit(root);
- }
-
- public IgniteRel root() {
- return root;
- }
-
- /** {@inheritDoc} */
- @Override
- public @Nullable SqlQueryType type() {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override
- public ResultSetMetadata metadata() {
- throw new IllegalStateException("Metadata is not available for a
FragmentPlan.");
- }
-
- /** {@inheritDoc} */
- @Override
- public QueryPlan copy() {
- return this;
- }
-}
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
index c78501322f..3c490600f7 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
@@ -161,7 +161,6 @@ public class StopCalciteModuleTest {
when(clusterSrvc.topologyService()).thenReturn(topologySrvc);
when(topologySrvc.localMember()).thenReturn(localNode);
when(topologySrvc.allMembers()).thenReturn(List.of(localNode));
- when(topologySrvc.getByConsistentId(any())).thenReturn(localNode);
SchemaDescriptor schemaDesc = new SchemaDescriptor(
1,
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 c2ab89e1d6..d9ee94036c 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
@@ -96,7 +96,6 @@ import
org.apache.ignite.internal.testframework.IgniteTestUtils.RunnableX;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.util.ArrayUtils;
import org.apache.ignite.lang.ErrorGroups.Sql;
-import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
@@ -233,19 +232,17 @@ public class ExecutionServiceImplTest {
testCluster.node(nodeNames.get(2)).interceptor((nodeName, msg,
original) -> {
if (msg instanceof QueryStartRequest) {
- try {
-
testCluster.node(nodeNames.get(2)).messageService().send(nodeName, new
SqlQueryMessagesFactory().queryStartResponse()
- .queryId(((QueryStartRequest) msg).queryId())
- .fragmentId(((QueryStartRequest) msg).fragmentId())
- .error(expectedEx)
- .build()
- );
- } catch (IgniteInternalCheckedException e) {
- throw new
IgniteInternalException(OPERATION_INTERRUPTED_ERR, e);
- }
+
testCluster.node(nodeNames.get(2)).messageService().send(nodeName, new
SqlQueryMessagesFactory().queryStartResponse()
+ .queryId(((QueryStartRequest) msg).queryId())
+ .fragmentId(((QueryStartRequest) msg).fragmentId())
+ .error(expectedEx)
+ .build()
+ );
} else {
original.onMessage(nodeName, msg);
}
+
+ return CompletableFuture.completedFuture(null);
});
InternalTransaction tx = new NoOpTransaction(nodeNames.get(0));
@@ -416,10 +413,12 @@ public class ExecutionServiceImplTest {
original.onMessage(senderNodeName, msg);
});
+
+ return CompletableFuture.completedFuture(null);
} else {
// On other nodes, simulate that the node has already gone.
- throw new IgniteInternalException(Sql.MESSAGE_SEND_ERR,
- "Connection refused to " + node.nodeName + ", message
" + msg);
+ return CompletableFuture.failedFuture(new
IgniteInternalException(Sql.INTERNAL_ERR,
+ "Connection refused to " + node.nodeName + ", message
" + msg));
}
}));
@@ -485,7 +484,6 @@ public class ExecutionServiceImplTest {
mock(DdlCommandHandler.class),
taskExecutor,
ArrayRowHandler.INSTANCE,
- exchangeService,
ctx -> node.implementor(ctx, mailboxRegistry, exchangeService)
);
@@ -538,7 +536,6 @@ public class ExecutionServiceImplTest {
class TestNode {
private final Map<Short, MessageListener> msgListeners = new
ConcurrentHashMap<>();
private final Queue<RunnableX> pending = new
LinkedBlockingQueue<>();
- private volatile boolean dead = false;
private volatile List<Object[]> dataset = List.of();
private volatile MessageInterceptor interceptor = null;
@@ -552,14 +549,6 @@ public class ExecutionServiceImplTest {
this.taskExecutor = taskExecutor;
}
- public void dead(boolean dead) {
- this.dead = dead;
- }
-
- public boolean dead() {
- return dead;
- }
-
public void dataset(List<Object[]> dataset) {
this.dataset = dataset;
}
@@ -597,16 +586,10 @@ public class ExecutionServiceImplTest {
return new MessageService() {
/** {@inheritDoc} */
@Override
- public void send(String nodeName, NetworkMessage msg) {
+ public CompletableFuture<Void> send(String nodeName,
NetworkMessage msg) {
TestNode node = nodes.get(nodeName);
- node.onReceive(TestNode.this.nodeName, msg);
- }
-
- /** {@inheritDoc} */
- @Override
- public boolean alive(String nodeName) {
- return !nodes.get(nodeName).dead();
+ return node.onReceive(TestNode.this.nodeName, msg);
}
/** {@inheritDoc} */
@@ -665,7 +648,7 @@ public class ExecutionServiceImplTest {
};
}
- private void onReceive(String senderNodeName, NetworkMessage
message) {
+ private CompletableFuture<Void> onReceive(String senderNodeName,
NetworkMessage message) {
MessageListener original = (nodeName, msg) -> {
MessageListener listener =
msgListeners.get(msg.messageType());
@@ -685,15 +668,18 @@ public class ExecutionServiceImplTest {
MessageInterceptor interceptor = this.interceptor;
if (interceptor != null) {
- interceptor.intercept(senderNodeName, message, original);
- } else {
- original.onMessage(senderNodeName, message);
+ return interceptor.intercept(senderNodeName, message,
original);
}
+
+ original.onMessage(senderNodeName, message);
+
+ return CompletableFuture.completedFuture(null);
}
}
+ @FunctionalInterface
interface MessageInterceptor {
- void intercept(String senderNodeName, NetworkMessage msg,
MessageListener original);
+ CompletableFuture<Void> intercept(String senderNodeName,
NetworkMessage msg, MessageListener original);
}
}
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 9678100c33..038cf54743 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
@@ -120,7 +120,6 @@ public class TestNode implements LifecycleAware {
mock(DdlCommandHandler.class),
taskExecutor,
rowHandler,
- exchangeService,
ctx -> new LogicalRelImplementor<Object[]>(
ctx,
new HashFunctionFactoryImpl<>(schemaManager,
rowHandler),
