You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ni...@apache.org on 2020/09/21 15:07:02 UTC
[ignite] branch master updated: IGNITE-13310 SQL queries tracing
(#8204)
This is an automated email from the ASF dual-hosted git repository.
nizhikov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e322d37 IGNITE-13310 SQL queries tracing (#8204)
e322d37 is described below
commit e322d375156adf9e84081bcfc510877603e3e883
Author: Mikhail Petrov <32...@users.noreply.github.com>
AuthorDate: Mon Sep 21 18:06:45 2020 +0300
IGNITE-13310 SQL queries tracing (#8204)
---
.../apache/ignite/util/GridCommandHandlerTest.java | 8 +-
...GridCommandHandlerTracingConfigurationTest.java | 5 +
.../processors/bulkload/BulkLoadProcessor.java | 37 +-
.../processors/query/GridRunningQueryInfo.java | 13 +
.../processors/query/RunningQueryManager.java | 38 +-
.../internal/processors/tracing/SpanTags.java | 21 +
.../internal/processors/tracing/SpanType.java | 77 ++-
.../processors/tracing/TraceableIterator.java | 69 +++
.../GridTracingConfigurationManager.java | 4 +
.../java/org/apache/ignite/spi/tracing/Scope.java | 5 +-
.../spi/tracing/TracingConfigurationManager.java | 10 +
.../internal/TestRecordingCommunicationSpi.java | 82 ++-
...ngeFreeCellularSwitchComplexOperationsTest.java | 2 +-
...ridExchangeFreeCellularSwitchIsolationTest.java | 6 +-
.../TxPartitionCounterStateAbstractTest.java | 3 +-
...ridCommandHandlerClusterByClassTest_help.output | 10 +-
...andHandlerClusterByClassWithSSLTest_help.output | 10 +-
.../cache/query/RegisteredQueryCursor.java | 63 +-
.../processors/query/h2/CommandProcessor.java | 2 +-
.../processors/query/h2/H2FieldsIterator.java | 7 +-
.../processors/query/h2/H2ResultSetIterator.java | 25 +-
.../processors/query/h2/IgniteH2Indexing.java | 158 +++--
.../internal/processors/query/h2/QueryParser.java | 11 +-
.../processors/query/h2/database/H2TreeIndex.java | 180 +++---
.../processors/query/h2/dml/DmlBatchSender.java | 35 +-
.../internal/processors/query/h2/dml/DmlUtils.java | 39 +-
.../query/h2/twostep/AbstractReducer.java | 49 +-
.../query/h2/twostep/GridMapQueryExecutor.java | 257 ++++----
.../query/h2/twostep/GridReduceQueryExecutor.java | 114 ++--
.../h2/twostep/PartitionReservationManager.java | 287 ++++-----
.../query/h2/twostep/ReduceIndexIterator.java | 17 +-
.../query/h2/twostep/msg/GridH2RowRange.java | 9 +-
modules/opencensus/pom.xml | 7 +
.../monitoring/opencensus/AbstractTracingTest.java | 4 +
.../opencensus/IgniteOpenCensusSuite.java | 4 +-
.../opencensus/OpenCensusSqlJdbcTracingTest.java | 168 ++++++
.../opencensus/OpenCensusSqlNativeTracingTest.java | 646 +++++++++++++++++++++
37 files changed, 1931 insertions(+), 551 deletions(-)
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index e45d9a1..a2ddf0b 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -96,7 +96,6 @@ import org.apache.ignite.internal.processors.cache.warmup.WarmUpTestPluginProvid
import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.G;
-import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.cache.VisorFindAndDeleteGarbageInPersistenceTaskResult;
@@ -1274,9 +1273,8 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
primSpi.waitForBlocked(clients.length);
// Unblock only finish messages from clients from 2 to 4.
- primSpi.stopBlock(true, new IgnitePredicate<T2<ClusterNode, GridIoMessage>>() {
- @Override public boolean apply(T2<ClusterNode, GridIoMessage> objects) {
- GridIoMessage iom = objects.get2();
+ primSpi.stopBlock(true, blockedMsg -> {
+ GridIoMessage iom = blockedMsg.ioMessage();
Message m = iom.message();
@@ -1288,7 +1286,7 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
return true;
}
- });
+ );
// Wait until queue is stable
for (Ignite ignite : G.allGrids()) {
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTracingConfigurationTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTracingConfigurationTest.java
index eb7abe2..985a25f 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTracingConfigurationTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTracingConfigurationTest.java
@@ -83,6 +83,11 @@ public class GridCommandHandlerTracingConfigurationTest extends GridCommandHandl
DFLT_CONFIG_MAP.put(
new TracingConfigurationCoordinates.Builder(Scope.DISCOVERY).build(),
TracingConfigurationManager.DEFAULT_DISCOVERY_CONFIGURATION);
+
+ DFLT_CONFIG_MAP.put(
+ new TracingConfigurationCoordinates.Builder(Scope.SQL).build(),
+ TracingConfigurationManager.DEFAULT_SQL_CONFIGURATION);
+
}
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
index a92bbdf..f391772 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/bulkload/BulkLoadProcessor.java
@@ -21,10 +21,18 @@ import java.util.List;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteIllegalStateException;
+import org.apache.ignite.internal.processors.query.GridRunningQueryInfo;
import org.apache.ignite.internal.processors.query.RunningQueryManager;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.NoopSpan;
+import org.apache.ignite.internal.processors.tracing.Span;
+import org.apache.ignite.internal.processors.tracing.Tracing;
import org.apache.ignite.internal.util.lang.IgniteClosureX;
import org.apache.ignite.lang.IgniteBiTuple;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_BATCH_PROCESS;
+
/**
* Bulk load (COPY) command processor used on server to keep various context data and process portions of input
* received from the client side.
@@ -54,6 +62,12 @@ public class BulkLoadProcessor implements AutoCloseable {
/** Exception, current load process ended with, or {@code null} if in progress or if succeded. */
private Exception failReason;
+ /** Tracing processor. */
+ private final Tracing tracing;
+
+ /** Span of the running query. */
+ private final Span qrySpan;
+
/**
* Creates bulk load processor.
*
@@ -63,14 +77,21 @@ public class BulkLoadProcessor implements AutoCloseable {
* @param outputStreamer Streamer that puts actual key/value into the cache.
* @param runningQryMgr Running query manager.
* @param qryId Running query id.
+ * @param tracing Tracing processor.
*/
public BulkLoadProcessor(BulkLoadParser inputParser, IgniteClosureX<List<?>, IgniteBiTuple<?, ?>> dataConverter,
- BulkLoadCacheWriter outputStreamer, RunningQueryManager runningQryMgr, Long qryId) {
+ BulkLoadCacheWriter outputStreamer, RunningQueryManager runningQryMgr, Long qryId, Tracing tracing) {
this.inputParser = inputParser;
this.dataConverter = dataConverter;
this.outputStreamer = outputStreamer;
this.runningQryMgr = runningQryMgr;
this.qryId = qryId;
+ this.tracing = tracing;
+
+ GridRunningQueryInfo qryInfo = runningQryMgr.runningQueryInfo(qryId);
+
+ qrySpan = qryInfo == null ? NoopSpan.INSTANCE : qryInfo.span();
+
isClosed = false;
}
@@ -91,15 +112,17 @@ public class BulkLoadProcessor implements AutoCloseable {
* @throws IgniteIllegalStateException when called after {@link #close()}.
*/
public void processBatch(byte[] batchData, boolean isLastBatch) throws IgniteCheckedException {
- if (isClosed)
- throw new IgniteIllegalStateException("Attempt to process a batch on a closed BulkLoadProcessor");
+ try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_BATCH_PROCESS, qrySpan))) {
+ if (isClosed)
+ throw new IgniteIllegalStateException("Attempt to process a batch on a closed BulkLoadProcessor");
- Iterable<List<Object>> inputRecords = inputParser.parseBatch(batchData, isLastBatch);
+ Iterable<List<Object>> inputRecords = inputParser.parseBatch(batchData, isLastBatch);
- for (List<Object> record : inputRecords) {
- IgniteBiTuple<?, ?> kv = dataConverter.apply(record);
+ for (List<Object> record : inputRecords) {
+ IgniteBiTuple<?, ?> kv = dataConverter.apply(record);
- outputStreamer.apply(kv);
+ outputStreamer.apply(kv);
+ }
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
index 1c82ee5..974d850 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/GridRunningQueryInfo.java
@@ -19,6 +19,8 @@ package org.apache.ignite.internal.processors.query;
import java.util.UUID;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.Span;
/**
* Query descriptor.
@@ -51,6 +53,9 @@ public class GridRunningQueryInfo {
/** */
private final QueryRunningFuture fut = new QueryRunningFuture();
+ /** Span of the running query. */
+ private final Span span;
+
/**
* Constructor.
*
@@ -81,6 +86,7 @@ public class GridRunningQueryInfo {
this.startTime = startTime;
this.cancel = cancel;
this.loc = loc;
+ this.span = MTC.span();
}
/**
@@ -169,4 +175,11 @@ public class GridRunningQueryInfo {
public UUID nodeId() {
return nodeId;
}
+
+ /**
+ * @return Span of the running query.
+ */
+ public Span span() {
+ return span;
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
index 9ba7938..8072378 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/RunningQueryManager.java
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.processors.cache.query.GridCacheQueryType;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.AtomicLongMetric;
import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric;
+import org.apache.ignite.internal.processors.tracing.Span;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.spi.systemview.view.SqlQueryHistoryView;
import org.apache.ignite.spi.systemview.view.SqlQueryView;
@@ -42,6 +43,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.SQL;
import static org.apache.ignite.internal.processors.cache.query.GridCacheQueryType.SQL_FIELDS;
import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
/**
* Keep information about all running queries.
@@ -173,24 +175,34 @@ public class RunningQueryManager {
if (qry == null)
return;
- //We need to collect query history and metrics only for SQL queries.
- if (isSqlQuery(qry)) {
- qry.runningFuture().onDone();
+ Span qrySpan = qry.span();
- qryHistTracker.collectHistory(qry, failed);
+ try {
+ if (failed)
+ qrySpan.addTag(ERROR, failReason::getMessage);
- if (!failed)
- successQrsCnt.increment();
- else {
- failedQrsCnt.increment();
+ //We need to collect query history and metrics only for SQL queries.
+ if (isSqlQuery(qry)) {
+ qry.runningFuture().onDone();
- // We measure cancel metric as "number of times user's queries ended up with query cancelled exception",
- // not "how many user's KILL QUERY command succeeded". These may be not the same if cancel was issued
- // right when query failed due to some other reason.
- if (QueryUtils.wasCancelled(failReason))
- canceledQrsCnt.increment();
+ qryHistTracker.collectHistory(qry, failed);
+
+ if (!failed)
+ successQrsCnt.increment();
+ else {
+ failedQrsCnt.increment();
+
+ // We measure cancel metric as "number of times user's queries ended up with query cancelled exception",
+ // not "how many user's KILL QUERY command succeeded". These may be not the same if cancel was issued
+ // right when query failed due to some other reason.
+ if (QueryUtils.wasCancelled(failReason))
+ canceledQrsCnt.increment();
+ }
}
}
+ finally {
+ qrySpan.end();
+ }
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
index d66b324..75c3a23 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanTags.java
@@ -84,6 +84,27 @@ public class SpanTags {
/** */
public static final String MESSAGE_CLASS = "message.class";
+ /** Text of SQL query. */
+ public static final String SQL_QRY_TEXT = "sql.query.text";
+
+ /** SQL schema. */
+ public static final String SQL_SCHEMA = "sql.schema";
+
+ /** Number of rows that result page contains. */
+ public static final String SQL_PAGE_ROWS = "sql.page.rows";
+
+ /** Number of rows that index range request contains. */
+ public static final String SQL_IDX_RANGE_ROWS = "sql.index.range.rows";
+
+ /** Name of SQL table. */
+ public static final String SQL_TABLE = "sql.table";
+
+ /** Name of SQL index. */
+ public static final String SQL_IDX = "sql.index";
+
+ /** Number of cache entries to be updated as a result of DML query. */
+ public static final String SQL_CACHE_UPDATES = "sql.cache.updates";
+
/** */
private SpanTags() {}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
index 6aa282e..693bb96 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/SpanType.java
@@ -159,7 +159,82 @@ public enum SpanType {
TX_NEAR_PREPARE_RESP(Scope.TX, "tx.near.process.prepare.response", 44),
/** Custom job call. */
- CUSTOM_JOB_CALL(Scope.COMMUNICATION, "job.call", 45, true);
+ CUSTOM_JOB_CALL(Scope.COMMUNICATION, "job.call", 45, true),
+
+ /** The overall execution of SQL query. */
+ SQL_QRY(Scope.SQL, "sql.query", 46, true),
+
+ /** Opening SQL query cursor. */
+ SQL_CURSOR_OPEN(Scope.SQL, "sql.cursor.open", 47),
+
+ /** Closing SQL query cursor. */
+ SQL_CURSOR_CLOSE(Scope.SQL, "sql.cursor.close", 48),
+
+ /** Cancellation SQL query cursor. */
+ SQL_CURSOR_CANCEL(Scope.SQL, "sql.cursor.cancel", 49),
+
+ /** Parsing SQL query. */
+ SQL_QRY_PARSE(Scope.SQL, "sql.query.parse", 50),
+
+ /** Processing SQL query execution request. */
+ SQL_QRY_EXEC_REQ(Scope.SQL, "sql.query.execute.request", 51),
+
+ /** Processing SQL next result page request. */
+ SQL_NEXT_PAGE_REQ(Scope.SQL, "sql.next.page.request", 52),
+
+ /** Processing mapped node response with requested SQL result page. */
+ SQL_PAGE_RESP(Scope.SQL, "sql.page.response", 53),
+
+ /** Execution SQL query by H2. */
+ SQL_QRY_EXECUTE(Scope.SQL, "sql.query.execute", 54),
+
+ /** Reading rows from cursor and preparing result page. */
+ SQL_PAGE_PREPARE(Scope.SQL, "sql.page.prepare", 55),
+
+ /** Processing SQL query fail response. */
+ SQL_FAIL_RESP(Scope.SQL, "sql.fail.response", 56),
+
+ /** Processing DML query request. */
+ SQL_DML_QRY_EXEC_REQ(Scope.SQL, "sql.dml.query.execute.request", 57),
+
+ /** Processing DML query response. */
+ SQL_DML_QRY_RESP(Scope.SQL, "sql.dml.query.response", 58),
+
+ /** Processing query cancellation request. */
+ SQL_QRY_CANCEL_REQ(Scope.SQL, "sql.query.cancel.request", 59),
+
+ /** Opening cursor iterator. */
+ SQL_ITER_OPEN(Scope.SQL, "sql.iterator.open", 60),
+
+ /** Opening cursor iterator. */
+ SQL_ITER_CLOSE(Scope.SQL, "sql.iterator.close", 61),
+
+ /** Fetching SQL query result page. */
+ SQL_PAGE_FETCH(Scope.SQL, "sql.page.fetch", 62),
+
+ /** Waiting for SQL query results page to be received. */
+ SQL_PAGE_WAIT(Scope.SQL, "sql.page.wait", 63),
+
+ /** Processing SQL index range request. */
+ SQL_IDX_RANGE_REQ(Scope.SQL, "sql.index.range.request", 64),
+
+ /** Processing SQL index range response. */
+ SQL_IDX_RANGE_RESP(Scope.SQL, "sql.index.range.response", 65),
+
+ /** Execution of SQL DML query. */
+ SQL_DML_QRY_EXECUTE(Scope.SQL, "sql.dml.query.execute", 66),
+
+ /** Execution of SQL command query which either DDL SQL queries or Ignite native SQL commands. */
+ SQL_CMD_QRY_EXECUTE(Scope.SQL, "sql.command.query.execute", 67),
+
+ /** SQL query partitions reservation. */
+ SQL_PARTITIONS_RESERVE(Scope.SQL, "sql.partitions.reserve", 68),
+
+ /** Update of cache as a result of the SQL DML query. */
+ SQL_CACHE_UPDATE(Scope.SQL, "sql.cache.update", 69),
+
+ /** Processing of incoming batch. */
+ SQL_BATCH_PROCESS(Scope.SQL, "sql.batch.process", 70);
/** Scope */
private Scope scope;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/TraceableIterator.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/TraceableIterator.java
new file mode 100644
index 0000000..6fed37e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/TraceableIterator.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.tracing;
+
+import java.util.Iterator;
+
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+
+/**
+ * Represents wrapper which allows the iterator methods to execute within context of a specified span.
+ */
+public class TraceableIterator<T> implements Iterator<T> {
+ /** Iterator to which all calls will be delegated. */
+ private final Iterator<T> iter;
+
+ /** Span that reperesents trace context in which iterator runs. */
+ private final Span span;
+
+ /**
+ * @param iter Iterator to which all calls will be delegated.
+ */
+ public TraceableIterator(Iterator<T> iter) {
+ this.iter = iter;
+ this.span = MTC.span();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean hasNext() {
+ MTC.supportInitial(span);
+
+ try {
+ return iter.hasNext();
+ }
+ catch (Throwable th) {
+ span.addTag(ERROR, th::getMessage);
+
+ throw th;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public T next() {
+ MTC.supportInitial(span);
+
+ try {
+ return iter.next();
+ }
+ catch (Throwable th) {
+ span.addTag(ERROR, th::getMessage);
+
+ throw th;
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/configuration/GridTracingConfigurationManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/configuration/GridTracingConfigurationManager.java
index cd03738..9ffd33f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/configuration/GridTracingConfigurationManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/configuration/GridTracingConfigurationManager.java
@@ -80,6 +80,10 @@ public class GridTracingConfigurationManager implements TracingConfigurationMana
new TracingConfigurationCoordinates.Builder(Scope.DISCOVERY).build(),
TracingConfigurationManager.DEFAULT_DISCOVERY_CONFIGURATION);
+ tmpDfltConfigurationMap.put(
+ new TracingConfigurationCoordinates.Builder(Scope.SQL).build(),
+ TracingConfigurationManager.DEFAULT_SQL_CONFIGURATION);
+
DEFAULT_CONFIGURATION_MAP = Collections.unmodifiableMap(tmpDfltConfigurationMap);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java b/modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java
index 3c5cc2c..0294183 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/tracing/Scope.java
@@ -31,7 +31,10 @@ public enum Scope {
COMMUNICATION((short)3),
/** Transactional scope. */
- TX((short)4);
+ TX((short)4),
+
+ /** SQL query scope. */
+ SQL((short)5);
/** Scope index. */
private final short idx;
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/tracing/TracingConfigurationManager.java b/modules/core/src/main/java/org/apache/ignite/spi/tracing/TracingConfigurationManager.java
index cfc2279..2915b42 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/tracing/TracingConfigurationManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/tracing/TracingConfigurationManager.java
@@ -29,6 +29,13 @@ import org.jetbrains.annotations.Nullable;
*/
@IgniteExperimental
public interface TracingConfigurationManager {
+ /** Default SQL tracing configuration. */
+ static final TracingConfigurationParameters DEFAULT_SQL_CONFIGURATION =
+ new TracingConfigurationParameters.Builder().
+ withSamplingRate(0d).
+ withIncludedScopes(Collections.emptySet()).
+ build();
+
/** Default transaction tracing configuration. */
static final TracingConfigurationParameters DEFAULT_TX_CONFIGURATION =
new TracingConfigurationParameters.Builder().
@@ -120,6 +127,9 @@ public interface TracingConfigurationManager {
return DEFAULT_COMMUNICATION_CONFIGURATION;
}
+ case SQL:
+ return DEFAULT_SQL_CONFIGURATION;
+
default: {
return NOOP_CONFIGURATION;
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
index e540c9a..8348147 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java
@@ -32,8 +32,10 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.managers.communication.GridIoMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Span;
import org.apache.ignite.internal.util.typedef.G;
-import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteBiPredicate;
@@ -60,7 +62,7 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
private List<Object> recordedMsgs = new ArrayList<>();
/** */
- private List<T2<ClusterNode, GridIoMessage>> blockedMsgs = new ArrayList<>();
+ private List<BlockedMessageDescriptor> blockedMsgs = new ArrayList<>();
/** */
private Map<Class<?>, Set<String>> blockCls = new HashMap<>();
@@ -121,7 +123,7 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
ignite.log().info("Block message [node=" + node.id() + ", order=" + node.order() +
", msg=" + ioMsg.message() + ']');
- blockedMsgs.add(new T2<>(node, ioMsg));
+ blockedMsgs.add(new BlockedMessageDescriptor(node, ioMsg, MTC.span()));
notifyAll();
@@ -255,9 +257,9 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
* @return {@code True} if has blocked message.
*/
private boolean hasMessage(Class<?> cls, String nodeName) {
- for (T2<ClusterNode, GridIoMessage> msg : blockedMsgs) {
- if (msg.get2().message().getClass() == cls &&
- nodeName.equals(msg.get1().attribute(ATTR_IGNITE_INSTANCE_NAME)))
+ for (BlockedMessageDescriptor blockedMsg : blockedMsgs) {
+ if (blockedMsg.ioMessage().message().getClass() == cls &&
+ nodeName.equals(blockedMsg.destinationNode().attribute(ATTR_IGNITE_INSTANCE_NAME)))
return true;
}
@@ -321,7 +323,7 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
* @param sndMsgs If {@code true} sends blocked messages.
* @param unblockPred If not null unblocks only messages allowed by predicate.
*/
- public void stopBlock(boolean sndMsgs, @Nullable IgnitePredicate<T2<ClusterNode, GridIoMessage>> unblockPred) {
+ public void stopBlock(boolean sndMsgs, @Nullable IgnitePredicate<BlockedMessageDescriptor> unblockPred) {
stopBlock(sndMsgs, unblockPred, true, true);
}
@@ -335,7 +337,7 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
* @param rmvBlockedMsgs {@code true} to remove blocked messages. Sometimes useful in conjunction with {@code
* sndMsgs=false}.
*/
- public void stopBlock(boolean sndMsgs, @Nullable IgnitePredicate<T2<ClusterNode, GridIoMessage>> unblockPred,
+ public void stopBlock(boolean sndMsgs, @Nullable IgnitePredicate<BlockedMessageDescriptor> unblockPred,
boolean clearFilters, boolean rmvBlockedMsgs) {
synchronized (this) {
if (clearFilters) {
@@ -343,25 +345,23 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
blockP = null;
}
- Iterator<T2<ClusterNode, GridIoMessage>> iter = blockedMsgs.iterator();
+ Iterator<BlockedMessageDescriptor> iter = blockedMsgs.iterator();
while (iter.hasNext()) {
- T2<ClusterNode, GridIoMessage> msg = iter.next();
+ BlockedMessageDescriptor blockedMsg = iter.next();
// It is important what predicate if called only once for each message.
- if (unblockPred != null && !unblockPred.apply(msg))
+ if (unblockPred != null && !unblockPred.apply(blockedMsg))
continue;
if (sndMsgs) {
- try {
- ignite.log().info("Send blocked message [node=" + msg.get1().id() +
- ", order=" + msg.get1().order() +
- ", msg=" + msg.get2().message() + ']');
+ try (TraceSurroundings ignored = MTC.supportContinual(blockedMsg.span())) {
+ ignite.log().info("Send blocked message " + blockedMsg);
- super.sendMessage(msg.get1(), msg.get2());
+ super.sendMessage(blockedMsg.destinationNode(), blockedMsg.ioMessage());
}
catch (Throwable e) {
- U.error(ignite.log(), "Failed to send blocked message: " + msg, e);
+ U.error(ignite.log(), "Failed to send blocked message: " + blockedMsg, e);
}
}
@@ -426,4 +426,52 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi {
};
}
+ /**
+ * Description of the blocked message.
+ */
+ public static class BlockedMessageDescriptor {
+ /** Destination node for the blocked message. */
+ private final ClusterNode destNode;
+
+ /** Blocked message. */
+ private final GridIoMessage msg;
+
+ /** Span in which context sending must be done. */
+ private final Span span;
+
+ /**
+ *
+ */
+ public BlockedMessageDescriptor(ClusterNode destNode, GridIoMessage msg, Span span) {
+ this.destNode = destNode;
+ this.msg = msg;
+ this.span = span;
+ }
+
+ /**
+ * @return Destination node for the blocked message.
+ */
+ public ClusterNode destinationNode() {
+ return destNode;
+ }
+
+ /**
+ * @return Blocked message.
+ */
+ public GridIoMessage ioMessage() {
+ return msg;
+ }
+
+ /**
+ * @return Span in which context sending must be done.
+ */
+ public Span span() {
+ return span;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return "[node=" + destNode.id() + ", order=" + destNode.order() + ", msg=" + msg.message() + ']';
+ }
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchComplexOperationsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchComplexOperationsTest.java
index 2bbcc84..1586361 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchComplexOperationsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchComplexOperationsTest.java
@@ -324,7 +324,7 @@ public class GridExchangeFreeCellularSwitchComplexOperationsTest extends GridExc
TestRecordingCommunicationSpi spi =
(TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
- spi.stopBlock(true, (t) -> true);
+ spi.stopBlock(true, blockedMsg -> true);
}
commitLatch.countDown();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchIsolationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchIsolationTest.java
index 42a4fb2..3b3af0b 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchIsolationTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridExchangeFreeCellularSwitchIsolationTest.java
@@ -355,8 +355,8 @@ public class GridExchangeFreeCellularSwitchIsolationTest extends GridExchangeFre
TestRecordingCommunicationSpi spi =
(TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
- spi.stopBlock(true, t -> {
- Message msg = t.get2().message();
+ spi.stopBlock(true, blockedMsg -> {
+ Message msg = blockedMsg.ioMessage().message();
return ((GridCacheTxRecoveryRequest)msg).nearXidVersion().equals(replTxVer.get());
});
@@ -400,7 +400,7 @@ public class GridExchangeFreeCellularSwitchIsolationTest extends GridExchangeFre
TestRecordingCommunicationSpi spi =
(TestRecordingCommunicationSpi)ignite.configuration().getCommunicationSpi();
- spi.stopBlock(true, (t) -> true);
+ spi.stopBlock(true, blockedMsg -> true);
}
partCommitLatch.countDown();
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
index d3745c2..9b7c61f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxPartitionCounterStateAbstractTest.java
@@ -517,7 +517,8 @@ public abstract class TxPartitionCounterStateAbstractTest extends GridCommonAbst
private GridFutureAdapter<?> createSendFuture(TestRecordingCommunicationSpi wrapperSpi, Message msg) {
GridFutureAdapter<?> fut = new GridFutureAdapter<>();
- fut.listen(fut1 -> wrapperSpi.stopBlock(true, objects -> objects.get2().message() == msg, false, true));
+ fut.listen(fut1 -> wrapperSpi.stopBlock(true, blockedMsg ->
+ blockedMsg.ioMessage().message() == msg, false, true));
return fut;
}
diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
index c0f5887..295905e 100644
--- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
+++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
@@ -151,19 +151,19 @@ If the file name isn't specified the output file name is: '<typeId>.bin'
control.(sh|bat) --tracing-configuration
Print tracing configuration:
- control.(sh|bat) --tracing-configuration get_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX]
+ control.(sh|bat) --tracing-configuration get_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL]
Print specific tracing configuration based on specified --scope and --label:
- control.(sh|bat) --tracing-configuration get (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX) [--label]
+ control.(sh|bat) --tracing-configuration get (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL) [--label]
Reset all specific tracing configuration the to default. If --scope is specified, then remove all label specific configuration for the given scope and reset given scope specific configuration to the default, if --scope is skipped then reset all tracing configurations to the default. Print tracing configuration.
- control.(sh|bat) --tracing-configuration reset_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX]
+ control.(sh|bat) --tracing-configuration reset_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL]
Reset specific tracing configuration to the default. If both --scope and --label are specified then remove given configuration, if only --scope is specified then reset given configuration to the default. Print reseted configuration.
- control.(sh|bat) --tracing-configuration reset (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX) [--label]
+ control.(sh|bat) --tracing-configuration reset (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL) [--label]
Set new tracing configuration. If both --scope and --label are specified then add or override label specific configuration, if only --scope is specified, then override scope specific configuration. Print applied configuration.
- control.(sh|bat) --tracing-configuration set (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX [--label] [--sampling-rate Decimal value between 0 and 1, where 0 means never and 1 means always. More or less reflects the probability of sampling specific trace.] [--included-scopes Set of scopes with comma as separator DISCOVERY|EXCHANGE|COMMUNICATION|TX])
+ control.(sh|bat) --tracing-configuration set (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL [--label] [--sampling-rate Decimal value between 0 and 1, where 0 means never and 1 means always. More or less reflects the probability of sampling specific trace.] [--included-scopes Set of scopes with comma as separator DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL])
Stop warm-up:
control.(sh|bat) --warm-up --stop
diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
index c0f5887..295905e 100644
--- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
+++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
@@ -151,19 +151,19 @@ If the file name isn't specified the output file name is: '<typeId>.bin'
control.(sh|bat) --tracing-configuration
Print tracing configuration:
- control.(sh|bat) --tracing-configuration get_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX]
+ control.(sh|bat) --tracing-configuration get_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL]
Print specific tracing configuration based on specified --scope and --label:
- control.(sh|bat) --tracing-configuration get (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX) [--label]
+ control.(sh|bat) --tracing-configuration get (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL) [--label]
Reset all specific tracing configuration the to default. If --scope is specified, then remove all label specific configuration for the given scope and reset given scope specific configuration to the default, if --scope is skipped then reset all tracing configurations to the default. Print tracing configuration.
- control.(sh|bat) --tracing-configuration reset_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX]
+ control.(sh|bat) --tracing-configuration reset_all [--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL]
Reset specific tracing configuration to the default. If both --scope and --label are specified then remove given configuration, if only --scope is specified then reset given configuration to the default. Print reseted configuration.
- control.(sh|bat) --tracing-configuration reset (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX) [--label]
+ control.(sh|bat) --tracing-configuration reset (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL) [--label]
Set new tracing configuration. If both --scope and --label are specified then add or override label specific configuration, if only --scope is specified, then override scope specific configuration. Print applied configuration.
- control.(sh|bat) --tracing-configuration set (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX [--label] [--sampling-rate Decimal value between 0 and 1, where 0 means never and 1 means always. More or less reflects the probability of sampling specific trace.] [--included-scopes Set of scopes with comma as separator DISCOVERY|EXCHANGE|COMMUNICATION|TX])
+ control.(sh|bat) --tracing-configuration set (--scope DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL [--label] [--sampling-rate Decimal value between 0 and 1, where 0 means never and 1 means always. More or less reflects the probability of sampling specific trace.] [--included-scopes Set of scopes with comma as separator DISCOVERY|EXCHANGE|COMMUNICATION|TX|SQL])
Stop warm-up:
control.(sh|bat) --warm-up --stop
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
index 0982c388..3945fec 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/cache/query/RegisteredQueryCursor.java
@@ -23,8 +23,19 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.internal.processors.cache.QueryCursorImpl;
import org.apache.ignite.internal.processors.query.GridQueryCancel;
+import org.apache.ignite.internal.processors.query.GridRunningQueryInfo;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.RunningQueryManager;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.NoopSpan;
+import org.apache.ignite.internal.processors.tracing.Span;
+import org.apache.ignite.internal.processors.tracing.TraceableIterator;
+import org.apache.ignite.internal.processors.tracing.Tracing;
+
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_CANCEL;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_CLOSE;
/**
* Query cursor for registered as running queries.
@@ -44,15 +55,22 @@ public class RegisteredQueryCursor<T> extends QueryCursorImpl<T> {
/** Exception caused query failed or {@code null} if it succeded. */
private Exception failReason;
+ /** Tracing processor. */
+ private final Tracing tracing;
+
+ /** Span of the running query. */
+ private final Span qrySpan;
+
/**
* @param iterExec Query executor.
* @param cancel Cancellation closure.
* @param runningQryMgr Running query manager.
* @param lazy Lazy mode flag.
* @param qryId Registered running query id.
+ * @param tracing Tracing processor.
*/
public RegisteredQueryCursor(Iterable<T> iterExec, GridQueryCancel cancel, RunningQueryManager runningQryMgr,
- boolean lazy, Long qryId) {
+ boolean lazy, Long qryId, Tracing tracing) {
super(iterExec, cancel, true, lazy);
assert runningQryMgr != null;
@@ -60,19 +78,25 @@ public class RegisteredQueryCursor<T> extends QueryCursorImpl<T> {
this.runningQryMgr = runningQryMgr;
this.qryId = qryId;
+ this.tracing = tracing;
+
+ GridRunningQueryInfo qryInfo = runningQryMgr.runningQueryInfo(qryId);
+
+ qrySpan = qryInfo == null ? NoopSpan.INSTANCE : qryInfo.span();
}
/** {@inheritDoc} */
@Override protected Iterator<T> iter() {
- try {
- if (lazy())
- return new RegisteredIterator(super.iter());
- else
- return super.iter();
+ try (TraceSurroundings ignored = MTC.supportContinual(qrySpan)) {
+ Iterator<T> iter = lazy() ? new RegisteredIterator(super.iter()) : super.iter();
+
+ return qrySpan != NoopSpan.INSTANCE ? new TraceableIterator<>(iter) : iter;
}
catch (Exception e) {
failReason = e;
+ qrySpan.addTag(ERROR, e::getMessage);
+
if (QueryUtils.wasCancelled(failReason))
unregisterQuery();
@@ -82,19 +106,36 @@ public class RegisteredQueryCursor<T> extends QueryCursorImpl<T> {
/** {@inheritDoc} */
@Override public void close() {
- super.close();
+ Span span = MTC.span();
- unregisterQuery();
+ try (
+ TraceSurroundings ignored = MTC.support(tracing.create(
+ SQL_CURSOR_CLOSE,
+ span != NoopSpan.INSTANCE ? span : qrySpan))
+ ) {
+ super.close();
+
+ unregisterQuery();
+ }
+ catch (Throwable th) {
+ qrySpan.addTag(ERROR, th::getMessage);
+
+ throw th;
+ }
}
/**
* Cancels query.
*/
public void cancel() {
- if (failReason == null)
- failReason = new QueryCancelledException();
+ try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_CURSOR_CANCEL, qrySpan))) {
+ if (failReason == null)
+ failReason = new QueryCancelledException();
+
+ qrySpan.addTag(ERROR, failReason::getMessage);
- close();
+ close();
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
index d85378a..e7135b6 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/CommandProcessor.java
@@ -1361,7 +1361,7 @@ public class CommandProcessor {
BulkLoadParser inputParser = BulkLoadParser.createParser(cmd.inputFormat());
BulkLoadProcessor processor = new BulkLoadProcessor(inputParser, dataConverter, outputWriter,
- idx.runningQueryManager(), qryId);
+ idx.runningQueryManager(), qryId, ctx.tracing());
BulkLoadAckClientParameters params = new BulkLoadAckClientParameters(cmd.localFileName(), cmd.packetSize());
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
index 3768af0..98deab2 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2FieldsIterator.java
@@ -24,6 +24,7 @@ import java.util.List;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
+import org.apache.ignite.internal.processors.tracing.Tracing;
/**
* Special field set iterator based on database result set.
@@ -43,6 +44,7 @@ public class H2FieldsIterator extends H2ResultSetIterator<List<?>> {
* @param mvccTracker Mvcc tracker.
* @param pageSize Page size.
* @param conn Connection.
+ * @param tracing Tracing processor.
* @throws IgniteCheckedException If failed.
*/
public H2FieldsIterator(
@@ -52,10 +54,11 @@ public class H2FieldsIterator extends H2ResultSetIterator<List<?>> {
int pageSize,
IgniteLogger log,
IgniteH2Indexing h2,
- H2QueryInfo qryInfo
+ H2QueryInfo qryInfo,
+ Tracing tracing
)
throws IgniteCheckedException {
- super(data, pageSize, log, h2, qryInfo);
+ super(data, pageSize, log, h2, qryInfo, tracing);
assert conn != null;
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
index 1473af0..3fba0b1 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2ResultSetIterator.java
@@ -30,6 +30,9 @@ import org.apache.ignite.cache.query.QueryCancelledException;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2Table;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2ValueCacheObject;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Tracing;
import org.apache.ignite.internal.util.lang.GridCloseableIterator;
import org.apache.ignite.internal.util.lang.GridIteratorAdapter;
import org.apache.ignite.internal.util.typedef.F;
@@ -40,6 +43,10 @@ import org.h2.jdbc.JdbcResultSet;
import org.h2.result.ResultInterface;
import org.h2.value.Value;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_PAGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_CLOSE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_FETCH;
+
/**
* Iterator over result set.
*/
@@ -100,11 +107,15 @@ public abstract class H2ResultSetIterator<T> extends GridIteratorAdapter<T> impl
/** Fetch size interceptor. */
final H2QueryFetchSizeInterceptor fetchSizeInterceptor;
+ /** Tracing processor. */
+ protected final Tracing tracing;
+
/**
* @param data Data array.
* @param log Logger.
* @param h2 Indexing H2.
* @param pageSize Page size.
+ * @param tracing Tracing processor.
* @throws IgniteCheckedException If failed.
*/
protected H2ResultSetIterator(
@@ -112,11 +123,13 @@ public abstract class H2ResultSetIterator<T> extends GridIteratorAdapter<T> impl
int pageSize,
IgniteLogger log,
IgniteH2Indexing h2,
- H2QueryInfo qryInfo
+ H2QueryInfo qryInfo,
+ Tracing tracing
)
throws IgniteCheckedException {
this.pageSize = pageSize;
this.data = data;
+ this.tracing = tracing;
try {
res = (ResultInterface)RESULT_FIELD.get(data);
@@ -158,7 +171,7 @@ public abstract class H2ResultSetIterator<T> extends GridIteratorAdapter<T> impl
private boolean fetchPage() throws IgniteCheckedException {
lockTables();
- try {
+ try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_PAGE_FETCH, MTC.span()))) {
GridH2Table.checkTablesVersions(ses);
page.clear();
@@ -198,6 +211,8 @@ public abstract class H2ResultSetIterator<T> extends GridIteratorAdapter<T> impl
}
}
+ MTC.span().addTag(SQL_PAGE_ROWS, () -> Integer.toString(page.size()));
+
if (F.isEmpty(page)) {
rowIter = null;
@@ -333,9 +348,11 @@ public abstract class H2ResultSetIterator<T> extends GridIteratorAdapter<T> impl
if (closed)
return;
- closed = true;
+ try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_ITER_CLOSE, MTC.span()))) {
+ closed = true;
- onClose();
+ onClose();
+ }
}
/** {@inheritDoc} */
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
index c8f2d65..de64068 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -151,6 +151,9 @@ import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryReq
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitor;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorClosure;
import org.apache.ignite.internal.processors.query.schema.SchemaIndexCacheVisitorImpl;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Span;
import org.apache.ignite.internal.sql.command.SqlCommand;
import org.apache.ignite.internal.sql.command.SqlCommitTransactionCommand;
import org.apache.ignite.internal.sql.command.SqlRollbackTransactionCommand;
@@ -209,6 +212,15 @@ import static org.apache.ignite.internal.processors.query.h2.H2Utils.generateFie
import static org.apache.ignite.internal.processors.query.h2.H2Utils.session;
import static org.apache.ignite.internal.processors.query.h2.H2Utils.validateTypeDescriptor;
import static org.apache.ignite.internal.processors.query.h2.H2Utils.zeroCursor;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_QRY_TEXT;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_SCHEMA;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CMD_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_DML_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_EXECUTE;
/**
* Indexing implementation based on H2 database engine. In this implementation main query language is SQL,
@@ -557,7 +569,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
@Override public GridCloseableIterator<List<?>> iterator() throws IgniteCheckedException {
H2PooledConnection conn = connections().connection(qryDesc.schemaName());
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_ITER_OPEN, MTC.span()))) {
H2Utils.setupConnection(conn, qctx,
qryDesc.distributedJoins(), qryDesc.enforceJoinOrder(), qryParams.lazy());
@@ -579,8 +591,16 @@ public class IgniteH2Indexing implements GridQueryIndexing {
qryInfo
);
- return new H2FieldsIterator(rs, mvccTracker, conn, qryParams.pageSize(),
- log, IgniteH2Indexing.this, qryInfo);
+ return new H2FieldsIterator(
+ rs,
+ mvccTracker,
+ conn,
+ qryParams.pageSize(),
+ log,
+ IgniteH2Indexing.this,
+ qryInfo,
+ ctx.tracing()
+ );
}
catch (IgniteCheckedException | RuntimeException | Error e) {
conn.close();
@@ -910,7 +930,11 @@ public class IgniteH2Indexing implements GridQueryIndexing {
enableDataPageScan(dataPageScanEnabled);
- try {
+ try (
+ TraceSurroundings ignored = MTC.support(ctx.tracing()
+ .create(SQL_QRY_EXECUTE, MTC.span())
+ .addTag(SQL_QRY_TEXT, () -> sql))
+ ) {
ResultSet rs = executeSqlQuery(conn, stmt, timeoutMillis, cancel);
if (qryInfo != null && qryInfo.time() > longRunningQryMgr.getTimeout())
@@ -1001,7 +1025,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
Exception failReason = null;
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_CMD_QRY_EXECUTE, MTC.span()))) {
res = cmdProc.runCommand(qryDesc.sql(), cmdNative, cmdH2, qryParams, cliCtx, qryId);
return res.cursor();
@@ -1063,73 +1087,85 @@ public class IgniteH2Indexing implements GridQueryIndexing {
SqlFieldsQuery remainingQry = qry;
while (remainingQry != null) {
- // Parse.
- QueryParserResult parseRes = parser.parse(schemaName, remainingQry, !failOnMultipleStmts);
+ Span qrySpan = ctx.tracing().create(SQL_QRY, MTC.span())
+ .addTag(SQL_SCHEMA, () -> schemaName);
- remainingQry = parseRes.remainingQuery();
+ try (TraceSurroundings ignored = MTC.supportContinual(qrySpan)) {
+ // Parse.
+ QueryParserResult parseRes = parser.parse(schemaName, remainingQry, !failOnMultipleStmts);
- // Get next command.
- QueryDescriptor newQryDesc = parseRes.queryDescriptor();
- QueryParameters newQryParams = parseRes.queryParameters();
+ qrySpan.addTag(SQL_QRY_TEXT, () -> parseRes.queryDescriptor().sql());
- // Check if there is enough parameters. Batched statements are not checked at this point
- // since they pass parameters differently.
- if (!newQryDesc.batched()) {
- int qryParamsCnt = F.isEmpty(newQryParams.arguments()) ? 0 : newQryParams.arguments().length;
+ remainingQry = parseRes.remainingQuery();
- if (qryParamsCnt < parseRes.parametersCount())
- throw new IgniteSQLException("Invalid number of query parameters [expected=" +
- parseRes.parametersCount() + ", actual=" + qryParamsCnt + ']');
- }
+ // Get next command.
+ QueryDescriptor newQryDesc = parseRes.queryDescriptor();
+ QueryParameters newQryParams = parseRes.queryParameters();
- // Check if cluster state is valid.
- checkClusterState(parseRes);
+ // Check if there is enough parameters. Batched statements are not checked at this point
+ // since they pass parameters differently.
+ if (!newQryDesc.batched()) {
+ int qryParamsCnt = F.isEmpty(newQryParams.arguments()) ? 0 : newQryParams.arguments().length;
- // Execute.
- if (parseRes.isCommand()) {
- QueryParserResultCommand cmd = parseRes.command();
+ if (qryParamsCnt < parseRes.parametersCount())
+ throw new IgniteSQLException("Invalid number of query parameters [expected=" +
+ parseRes.parametersCount() + ", actual=" + qryParamsCnt + ']');
+ }
- assert cmd != null;
+ // Check if cluster state is valid.
+ checkClusterState(parseRes);
- FieldsQueryCursor<List<?>> cmdRes = executeCommand(
- newQryDesc,
- newQryParams,
- cliCtx,
- cmd
- );
+ // Execute.
+ if (parseRes.isCommand()) {
+ QueryParserResultCommand cmd = parseRes.command();
- res.add(cmdRes);
- }
- else if (parseRes.isDml()) {
- QueryParserResultDml dml = parseRes.dml();
+ assert cmd != null;
- assert dml != null;
+ FieldsQueryCursor<List<?>> cmdRes = executeCommand(
+ newQryDesc,
+ newQryParams,
+ cliCtx,
+ cmd
+ );
- List<? extends FieldsQueryCursor<List<?>>> dmlRes = executeDml(
- newQryDesc,
- newQryParams,
- dml,
- cancel
- );
+ res.add(cmdRes);
+ }
+ else if (parseRes.isDml()) {
+ QueryParserResultDml dml = parseRes.dml();
- res.addAll(dmlRes);
- }
- else {
- assert parseRes.isSelect();
+ assert dml != null;
- QueryParserResultSelect select = parseRes.select();
+ List<? extends FieldsQueryCursor<List<?>>> dmlRes = executeDml(
+ newQryDesc,
+ newQryParams,
+ dml,
+ cancel
+ );
- assert select != null;
+ res.addAll(dmlRes);
+ }
+ else {
+ assert parseRes.isSelect();
- List<? extends FieldsQueryCursor<List<?>>> qryRes = executeSelect(
- newQryDesc,
- newQryParams,
- select,
- keepBinary,
- cancel
- );
+ QueryParserResultSelect select = parseRes.select();
+
+ assert select != null;
- res.addAll(qryRes);
+ List<? extends FieldsQueryCursor<List<?>>> qryRes = executeSelect(
+ newQryDesc,
+ newQryParams,
+ select,
+ keepBinary,
+ cancel
+ );
+
+ res.addAll(qryRes);
+ }
+ }
+ catch (Throwable th) {
+ qrySpan.addTag(ERROR, th::getMessage).end();
+
+ throw th;
}
}
@@ -1170,7 +1206,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
Exception failReason = null;
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_DML_QRY_EXECUTE, MTC.span()))) {
if (!dml.mvccEnabled() && !updateInTxAllowed && ctx.cache().context().tm().inUserTx()) {
throw new IgniteSQLException("DML statements are not allowed inside a transaction over " +
"cache(s) with TRANSACTIONAL atomicity mode (change atomicity mode to " +
@@ -1253,7 +1289,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
// Register query.
Long qryId = registerRunningQuery(qryDesc, cancel);
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_CURSOR_OPEN, MTC.span()))) {
GridNearTxLocal tx = null;
MvccQueryTracker tracker = null;
GridCacheContext mvccCctx = null;
@@ -1298,8 +1334,8 @@ public class IgniteH2Indexing implements GridQueryIndexing {
if (select.forUpdate() && inTx)
iter = lockSelectedRows(iter, mvccCctx, timeout, qryParams.pageSize());
- RegisteredQueryCursor<List<?>> cursor =
- new RegisteredQueryCursor<>(iter, cancel, runningQueryManager(), qryParams.lazy(), qryId);
+ RegisteredQueryCursor<List<?>> cursor = new RegisteredQueryCursor<>(iter, cancel, runningQueryManager(),
+ qryParams.lazy(), qryId, ctx.tracing());
cancel.add(cursor::cancel);
@@ -1711,7 +1747,7 @@ public class IgniteH2Indexing implements GridQueryIndexing {
iter = new Iterable<List<?>>() {
@SuppressWarnings("NullableProblems")
@Override public Iterator<List<?>> iterator() {
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_ITER_OPEN, MTC.span()))) {
return IgniteH2Indexing.this.rdcQryExec.query(
qryDesc.schemaName(),
twoStepQry,
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
index 08691c0..a7979cd 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/QueryParser.java
@@ -54,6 +54,8 @@ import org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlSelect;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlStatement;
import org.apache.ignite.internal.processors.query.h2.sql.GridSqlTable;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.sql.SqlParseException;
import org.apache.ignite.internal.sql.SqlParser;
import org.apache.ignite.internal.sql.SqlStrictParseException;
@@ -82,6 +84,7 @@ import org.h2.command.Prepared;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter.keyColumn;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_PARSE;
/**
* Parser module. Splits incoming request into a series of parsed results.
@@ -135,11 +138,13 @@ public class QueryParser {
* @return Parsing result that contains Parsed leading query and remaining sql script.
*/
public QueryParserResult parse(String schemaName, SqlFieldsQuery qry, boolean remainingAllowed) {
- QueryParserResult res = parse0(schemaName, qry, remainingAllowed);
+ try (TraceSurroundings ignored = MTC.support(idx.kernalContext().tracing().create(SQL_QRY_PARSE, MTC.span()))) {
+ QueryParserResult res = parse0(schemaName, qry, remainingAllowed);
- checkQueryType(qry, res.isSelect());
+ checkQueryType(qry, res.isSelect());
- return res;
+ return res;
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
index b2d770e..f28ec17 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
@@ -75,6 +75,9 @@ import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2RowRange
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2RowRangeBounds;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2ValueMessage;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2ValueMessageFactory;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Span;
import org.apache.ignite.internal.util.GridSpinBusyLock;
import org.apache.ignite.internal.util.IgniteTree;
import org.apache.ignite.internal.util.lang.GridCursor;
@@ -105,6 +108,12 @@ import static org.apache.ignite.internal.metric.IoStatisticsType.SORTED_INDEX;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2IndexRangeResponse.STATUS_ERROR;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2IndexRangeResponse.STATUS_NOT_FOUND;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2IndexRangeResponse.STATUS_OK;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_IDX;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_IDX_RANGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_TABLE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_IDX_RANGE_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_IDX_RANGE_RESP;
import static org.h2.result.Row.MEMORY_CALCULATE;
/**
@@ -732,85 +741,108 @@ public class H2TreeIndex extends H2TreeIndexBase {
* @param msg Request message.
*/
private void onIndexRangeRequest(final ClusterNode node, final GridH2IndexRangeRequest msg) {
- GridH2IndexRangeResponse res = new GridH2IndexRangeResponse();
-
- res.originNodeId(msg.originNodeId());
- res.queryId(msg.queryId());
- res.originSegmentId(msg.originSegmentId());
- res.segment(msg.segment());
- res.batchLookupId(msg.batchLookupId());
-
- QueryContext qctx = qryCtxRegistry.getShared(
- msg.originNodeId(),
- msg.queryId(),
- msg.originSegmentId()
- );
+ // We don't use try with resources on purpose - the catch block must also be executed in the context of this span.
+ TraceSurroundings trace = MTC.support(ctx.tracing().create(SQL_IDX_RANGE_REQ, MTC.span()));
- if (qctx == null)
- res.status(STATUS_NOT_FOUND);
- else {
- DistributedJoinContext joinCtx = qctx.distributedJoinContext();
+ Span span = MTC.span();
- assert joinCtx != null;
+ try {
+ span.addTag(SQL_IDX, () -> idxName);
+ span.addTag(SQL_TABLE, () -> tblName);
- try {
- RangeSource src;
+ GridH2IndexRangeResponse res = new GridH2IndexRangeResponse();
- if (msg.bounds() != null) {
- // This is the first request containing all the search rows.
- assert !msg.bounds().isEmpty() : "empty bounds";
+ res.originNodeId(msg.originNodeId());
+ res.queryId(msg.queryId());
+ res.originSegmentId(msg.originSegmentId());
+ res.segment(msg.segment());
+ res.batchLookupId(msg.batchLookupId());
- src = new RangeSource(this, msg.bounds(), msg.segment(), filter(qctx));
- }
- else {
- // This is request to fetch next portion of data.
- src = joinCtx.getSource(node.id(), msg.segment(), msg.batchLookupId());
+ QueryContext qctx = qryCtxRegistry.getShared(
+ msg.originNodeId(),
+ msg.queryId(),
+ msg.originSegmentId()
+ );
- assert src != null;
- }
+ if (qctx == null)
+ res.status(STATUS_NOT_FOUND);
+ else {
+ DistributedJoinContext joinCtx = qctx.distributedJoinContext();
+
+ assert joinCtx != null;
- List<GridH2RowRange> ranges = new ArrayList<>();
+ try {
+ RangeSource src;
- int maxRows = joinCtx.pageSize();
+ if (msg.bounds() != null) {
+ // This is the first request containing all the search rows.
+ assert !msg.bounds().isEmpty() : "empty bounds";
- assert maxRows > 0 : maxRows;
+ src = new RangeSource(this, msg.bounds(), msg.segment(), filter(qctx));
+ }
+ else {
+ // This is request to fetch next portion of data.
+ src = joinCtx.getSource(node.id(), msg.segment(), msg.batchLookupId());
- while (maxRows > 0) {
- GridH2RowRange range = src.next(maxRows);
+ assert src != null;
+ }
- if (range == null)
- break;
+ List<GridH2RowRange> ranges = new ArrayList<>();
- ranges.add(range);
+ int maxRows = joinCtx.pageSize();
- if (range.rows() != null)
- maxRows -= range.rows().size();
- }
+ assert maxRows > 0 : maxRows;
- assert !ranges.isEmpty();
+ while (maxRows > 0) {
+ GridH2RowRange range = src.next(maxRows);
- if (src.hasMoreRows()) {
- // Save source for future fetches.
- if (msg.bounds() != null)
- joinCtx.putSource(node.id(), msg.segment(), msg.batchLookupId(), src);
- }
- else if (msg.bounds() == null) {
- // Drop saved source.
- joinCtx.putSource(node.id(), msg.segment(), msg.batchLookupId(), null);
+ if (range == null)
+ break;
+
+ ranges.add(range);
+
+ if (range.rows() != null)
+ maxRows -= range.rows().size();
+ }
+
+ assert !ranges.isEmpty();
+
+ if (src.hasMoreRows()) {
+ // Save source for future fetches.
+ if (msg.bounds() != null)
+ joinCtx.putSource(node.id(), msg.segment(), msg.batchLookupId(), src);
+ }
+ else if (msg.bounds() == null) {
+ // Drop saved source.
+ joinCtx.putSource(node.id(), msg.segment(), msg.batchLookupId(), null);
+ }
+
+ res.ranges(ranges);
+ res.status(STATUS_OK);
+
+ span.addTag(SQL_IDX_RANGE_ROWS, () ->
+ Integer.toString(ranges.stream().mapToInt(GridH2RowRange::rowsSize).sum()));
}
+ catch (Throwable th) {
+ span.addTag(ERROR, th::getMessage);
- res.ranges(ranges);
- res.status(STATUS_OK);
- }
- catch (Throwable th) {
- U.error(log, "Failed to process request: " + msg, th);
+ U.error(log, "Failed to process request: " + msg, th);
- res.error(th.getClass() + ": " + th.getMessage());
- res.status(STATUS_ERROR);
+ res.error(th.getClass() + ": " + th.getMessage());
+ res.status(STATUS_ERROR);
+ }
}
+
+ send(singletonList(node), res);
}
+ catch (Throwable th) {
+ span.addTag(ERROR, th::getMessage);
- send(singletonList(node), res);
+ throw th;
+ }
+ finally {
+ trace.close();
+ }
}
/**
@@ -818,29 +850,31 @@ public class H2TreeIndex extends H2TreeIndexBase {
* @param msg Response message.
*/
private void onIndexRangeResponse(ClusterNode node, GridH2IndexRangeResponse msg) {
- QueryContext qctx = qryCtxRegistry.getShared(
- msg.originNodeId(),
- msg.queryId(),
- msg.originSegmentId()
- );
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_IDX_RANGE_RESP, MTC.span()))) {
+ QueryContext qctx = qryCtxRegistry.getShared(
+ msg.originNodeId(),
+ msg.queryId(),
+ msg.originSegmentId()
+ );
- if (qctx == null)
- return;
+ if (qctx == null)
+ return;
- DistributedJoinContext joinCtx = qctx.distributedJoinContext();
+ DistributedJoinContext joinCtx = qctx.distributedJoinContext();
- assert joinCtx != null;
+ assert joinCtx != null;
- Map<SegmentKey, RangeStream> streams = joinCtx.getStreams(msg.batchLookupId());
+ Map<SegmentKey, RangeStream> streams = joinCtx.getStreams(msg.batchLookupId());
- if (streams == null)
- return;
+ if (streams == null)
+ return;
- RangeStream stream = streams.get(new SegmentKey(node, msg.segment()));
+ RangeStream stream = streams.get(new SegmentKey(node, msg.segment()));
- assert stream != null;
+ assert stream != null;
- stream.onResponse(msg);
+ stream.onResponse(msg);
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlBatchSender.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlBatchSender.java
index dff60b3..874a44b 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlBatchSender.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlBatchSender.java
@@ -42,10 +42,15 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteClusterReadOnlyException;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.odbc.SqlStateCode;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.createJdbcSqlException;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_CACHE_UPDATES;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CACHE_UPDATE;
/**
* Batch sender class.
@@ -193,22 +198,30 @@ public class DmlBatchSender {
* @param batch Batch.
*/
private void sendBatch(Batch batch) {
- DmlPageProcessingResult pageRes = processPage(cctx, batch);
+ try (
+ TraceSurroundings ignored = MTC.support(cctx.kernalContext().tracing()
+ .create(SQL_CACHE_UPDATE, MTC.span())
+ .addTag(SQL_CACHE_UPDATES, () -> Integer.toString(batch.size())))
+ ) {
+ DmlPageProcessingResult pageRes = processPage(cctx, batch);
- batch.clear();
+ batch.clear();
- updateCnt += pageRes.count();
+ updateCnt += pageRes.count();
- if (failedKeys == null)
- failedKeys = new ArrayList<>();
+ if (failedKeys == null)
+ failedKeys = new ArrayList<>();
- failedKeys.addAll(F.asList(pageRes.errorKeys()));
+ failedKeys.addAll(F.asList(pageRes.errorKeys()));
- if (pageRes.error() != null) {
- if (err == null)
- err = pageRes.error();
- else
- err.setNextException(pageRes.error());
+ if (pageRes.error() != null) {
+ MTC.span().addTag(ERROR, pageRes.error()::getMessage);
+
+ if (err == null)
+ err = pageRes.error();
+ else
+ err.setNextException(pageRes.error());
+ }
}
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
index 3fdb945..4152297 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/dml/DmlUtils.java
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.processors.query.h2.DmlStatementsProcessor;
import org.apache.ignite.internal.processors.query.h2.H2Utils;
import org.apache.ignite.internal.processors.query.h2.UpdateResult;
import org.apache.ignite.internal.processors.query.h2.opt.GridH2RowDescriptor;
+import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T3;
import org.apache.ignite.internal.util.typedef.X;
@@ -60,6 +61,8 @@ import org.h2.value.ValueTimestamp;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.DUPLICATE_KEY;
import static org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode.createJdbcSqlException;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_CACHE_UPDATES;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CACHE_UPDATE;
/**
* DML utility methods.
@@ -194,10 +197,16 @@ public class DmlUtils {
if (plan.rowCount() == 1) {
IgniteBiTuple t = plan.processRow(cursor.iterator().next());
- if (cctx.cache().putIfAbsent(t.getKey(), t.getValue()))
- return 1;
- else
- throw new TransactionDuplicateKeyException("Duplicate key during INSERT [key=" + t.getKey() + ']');
+ try (
+ MTC.TraceSurroundings ignored = MTC.support(cctx.kernalContext().tracing()
+ .create(SQL_CACHE_UPDATE, MTC.span())
+ .addTag(SQL_CACHE_UPDATES, () -> "1"))
+ ) {
+ if (cctx.cache().putIfAbsent(t.getKey(), t.getValue()))
+ return 1;
+ else
+ throw new TransactionDuplicateKeyException("Duplicate key during INSERT [key=" + t.getKey() + ']');
+ }
}
else {
// Keys that failed to INSERT due to duplication.
@@ -299,7 +308,13 @@ public class DmlUtils {
if (plan.rowCount() == 1) {
IgniteBiTuple t = plan.processRow(cursor.iterator().next());
- cctx.cache().put(t.getKey(), t.getValue());
+ try (
+ MTC.TraceSurroundings ignored = MTC.support(cctx.kernalContext().tracing()
+ .create(SQL_CACHE_UPDATE, MTC.span())
+ .addTag(SQL_CACHE_UPDATES, () -> "1"))
+ ) {
+ cctx.cache().put(t.getKey(), t.getValue());
+ }
return 1;
}
@@ -316,12 +331,18 @@ public class DmlUtils {
rows.put(t.getKey(), t.getValue());
if ((pageSize > 0 && rows.size() == pageSize) || !it.hasNext()) {
- cctx.cache().putAll(rows);
+ try (
+ MTC.TraceSurroundings ignored = MTC.support(cctx.kernalContext().tracing()
+ .create(SQL_CACHE_UPDATE, MTC.span())
+ .addTag(SQL_CACHE_UPDATES, () -> Integer.toString(rows.size())))
+ ) {
+ cctx.cache().putAll(rows);
- resCnt += rows.size();
+ resCnt += rows.size();
- if (it.hasNext())
- rows.clear();
+ if (it.hasNext())
+ rows.clear();
+ }
}
}
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/AbstractReducer.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/AbstractReducer.java
index 3b08fe1..c1e32b8 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/AbstractReducer.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/AbstractReducer.java
@@ -32,6 +32,8 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageResponse;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.h2.index.Cursor;
@@ -45,6 +47,9 @@ import static java.util.Objects.requireNonNull;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_MERGE_TABLE_MAX_SIZE;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SQL_MERGE_TABLE_PREFETCH_SIZE;
import static org.apache.ignite.IgniteSystemProperties.getInteger;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_PAGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_FETCH;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_WAIT;
/**
* Base class for reducer of remote index lookup results.
@@ -315,15 +320,19 @@ public abstract class AbstractReducer implements Reducer {
*/
protected final Iterator<Value[]> pollNextIterator(Pollable<ReduceResultPage> queue, Iterator<Value[]> iter) {
if (!iter.hasNext()) {
- ReduceResultPage page = takeNextPage(queue);
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_PAGE_FETCH, MTC.span()))) {
+ ReduceResultPage page = takeNextPage(queue);
- if (!page.isLast())
- page.fetchNextPage(); // Failed will throw an exception here.
+ if (!page.isLast())
+ page.fetchNextPage(); // Failed will throw an exception here.
- iter = page.rows();
+ iter = page.rows();
- // The received iterator must be empty in the dummy last page or on failure.
- assert iter.hasNext() || page.isDummyLast() || page.isFail();
+ MTC.span().addTag(SQL_PAGE_ROWS, () -> Integer.toString(page.rowsInPage()));
+
+ // The received iterator must be empty in the dummy last page or on failure.
+ assert iter.hasNext() || page.isDummyLast() || page.isFail();
+ }
}
return iter;
@@ -334,23 +343,25 @@ public abstract class AbstractReducer implements Reducer {
* @return Next page.
*/
private ReduceResultPage takeNextPage(Pollable<ReduceResultPage> queue) {
- ReduceResultPage page;
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_PAGE_WAIT, MTC.span()))) {
+ ReduceResultPage page;
- for (;;) {
- try {
- page = queue.poll(500, TimeUnit.MILLISECONDS);
- }
- catch (InterruptedException e) {
- throw new CacheException("Query execution was interrupted.", e);
- }
+ for (;;) {
+ try {
+ page = queue.poll(500, TimeUnit.MILLISECONDS);
+ }
+ catch (InterruptedException e) {
+ throw new CacheException("Query execution was interrupted.", e);
+ }
- if (page != null)
- break;
+ if (page != null)
+ break;
- checkSourceNodesAlive();
- }
+ checkSourceNodesAlive();
+ }
- return page;
+ return page;
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
index 0a9a8ba..06dc6d3 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridMapQueryExecutor.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import javax.cache.CacheException;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
@@ -66,6 +67,10 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Span;
+import org.apache.ignite.internal.processors.tracing.SpanType;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -81,6 +86,13 @@ import static org.apache.ignite.events.EventType.EVT_CACHE_QUERY_EXECUTED;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.QUERY_POOL;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest.isDataPageScanEnabled;
import static org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2ValueMessageFactory.toMessages;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_PAGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_QRY_TEXT;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_NEXT_PAGE_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_PREPARE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_CANCEL_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_EXEC_REQ;
/**
* Map query executor.
@@ -146,19 +158,21 @@ public class GridMapQueryExecutor {
* @param msg Message.
*/
public void onCancel(ClusterNode node, GridQueryCancelRequest msg) {
- long qryReqId = msg.queryRequestId();
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_QRY_CANCEL_REQ, MTC.span()))) {
+ long qryReqId = msg.queryRequestId();
- MapNodeResults nodeRess = resultsForNode(node.id());
+ MapNodeResults nodeRess = resultsForNode(node.id());
- boolean clear = qryCtxRegistry.clearShared(node.id(), qryReqId);
+ boolean clear = qryCtxRegistry.clearShared(node.id(), qryReqId);
- if (!clear) {
- nodeRess.onCancel(qryReqId);
+ if (!clear) {
+ nodeRess.onCancel(qryReqId);
- qryCtxRegistry.clearShared(node.id(), qryReqId);
- }
+ qryCtxRegistry.clearShared(node.id(), qryReqId);
+ }
- nodeRess.cancelRequest(qryReqId);
+ nodeRess.cancelRequest(qryReqId);
+ }
}
/**
@@ -216,28 +230,32 @@ public class GridMapQueryExecutor {
final int segment = i;
+ Span span = MTC.span();
+
ctx.closure().callLocal(
(Callable<Void>)() -> {
- onQueryRequest0(node,
- req.requestId(),
- segment,
- req.schemaName(),
- req.queries(),
- cacheIds,
- req.topologyVersion(),
- partsMap,
- parts,
- req.pageSize(),
- distributedJoins,
- enforceJoinOrder,
- false,
- timeout,
- params,
- lazy,
- req.mvccSnapshot(),
- dataPageScanEnabled);
-
- return null;
+ try (TraceSurroundings ignored = MTC.supportContinual(span)) {
+ onQueryRequest0(node,
+ req.requestId(),
+ segment,
+ req.schemaName(),
+ req.queries(),
+ cacheIds,
+ req.topologyVersion(),
+ partsMap,
+ parts,
+ req.pageSize(),
+ distributedJoins,
+ enforceJoinOrder,
+ false,
+ timeout,
+ params,
+ lazy,
+ req.mvccSnapshot(),
+ dataPageScanEnabled);
+
+ return null;
+ }
},
QUERY_POOL);
}
@@ -313,6 +331,12 @@ public class GridMapQueryExecutor {
QueryContext qctx = null;
+ // We don't use try with resources on purpose - the catch block must also be executed in the context of this span.
+ TraceSurroundings trace = MTC.support(ctx.tracing()
+ .create(SQL_QRY_EXEC_REQ, MTC.span())
+ .addTag(SQL_QRY_TEXT, () ->
+ qrys.stream().map(GridCacheSqlQuery::query).collect(Collectors.joining("; "))));
+
try {
if (topVer != null) {
// Reserve primary for topology version or explicit partitions.
@@ -528,6 +552,8 @@ public class GridMapQueryExecutor {
finally {
if (reserved != null)
reserved.release();
+
+ trace.close();
}
}
@@ -568,6 +594,11 @@ public class GridMapQueryExecutor {
MapNodeResults nodeResults = resultsForNode(node.id());
+ // We don't use try with resources on purpose - the catch block must also be executed in the context of this span.
+ TraceSurroundings trace = MTC.support(ctx.tracing()
+ .create(SpanType.SQL_DML_QRY_EXEC_REQ, MTC.span())
+ .addTag(SQL_QRY_TEXT, req::query));
+
try {
reserved = h2.partitionReservationManager().reservePartitions(
cacheIds,
@@ -641,6 +672,8 @@ public class GridMapQueryExecutor {
sendUpdateResponse(node, reqId, updRes, null);
}
catch (Exception e) {
+ MTC.span().addTag(ERROR, e::getMessage);
+
U.error(log, "Error processing dml request. [localNodeId=" + ctx.localNodeId() +
", nodeId=" + node.id() + ", req=" + req + ']', e);
@@ -651,6 +684,8 @@ public class GridMapQueryExecutor {
reserved.release();
nodeResults.removeUpdate(reqId);
+
+ trace.close();
}
}
@@ -669,6 +704,8 @@ public class GridMapQueryExecutor {
*/
private void sendError(ClusterNode node, long qryReqId, Throwable err) {
try {
+ MTC.span().addTag(ERROR, err::getMessage);
+
GridQueryFailResponse msg = new GridQueryFailResponse(qryReqId, err);
if (node.isLocal()) {
@@ -731,78 +768,80 @@ public class GridMapQueryExecutor {
* @param req Request.
*/
public void onNextPageRequest(final ClusterNode node, final GridQueryNextPageRequest req) {
- long reqId = req.queryRequestId();
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_NEXT_PAGE_REQ, MTC.span()))) {
+ long reqId = req.queryRequestId();
- final MapNodeResults nodeRess = qryRess.get(node.id());
+ final MapNodeResults nodeRess = qryRess.get(node.id());
- if (nodeRess == null) {
- sendError(node, reqId, new CacheException("No node result found for request: " + req));
+ if (nodeRess == null) {
+ sendError(node, reqId, new CacheException("No node result found for request: " + req));
- return;
- }
- else if (nodeRess.cancelled(reqId)) {
- sendQueryCancel(node, reqId);
+ return;
+ }
+ else if (nodeRess.cancelled(reqId)) {
+ sendQueryCancel(node, reqId);
- return;
- }
+ return;
+ }
- final MapQueryResults qryResults = nodeRess.get(reqId, req.segmentId());
+ final MapQueryResults qryResults = nodeRess.get(reqId, req.segmentId());
- if (qryResults == null)
- sendError(node, reqId, new CacheException("No query result found for request: " + req));
- else if (qryResults.cancelled())
- sendQueryCancel(node, reqId);
- else {
- try {
- MapQueryResult res = qryResults.result(req.query());
+ if (qryResults == null)
+ sendError(node, reqId, new CacheException("No query result found for request: " + req));
+ else if (qryResults.cancelled())
+ sendQueryCancel(node, reqId);
+ else {
+ try {
+ MapQueryResult res = qryResults.result(req.query());
- assert res != null;
+ assert res != null;
- try {
- // Session isn't set for lazy=false queries.
- // Also session == null when result already closed.
- res.lock();
- res.lockTables();
- res.checkTablesVersions();
-
- Boolean dataPageScanEnabled = isDataPageScanEnabled(req.getFlags());
-
- GridQueryNextPageResponse msg = prepareNextPage(
- nodeRess,
- node,
- qryResults,
- req.query(),
- req.segmentId(),
- req.pageSize(),
- dataPageScanEnabled);
-
- if (msg != null)
- sendNextPage(node, msg);
- }
- finally {
try {
- res.unlockTables();
+ // Session isn't set for lazy=false queries.
+ // Also session == null when result already closed.
+ res.lock();
+ res.lockTables();
+ res.checkTablesVersions();
+
+ Boolean dataPageScanEnabled = isDataPageScanEnabled(req.getFlags());
+
+ GridQueryNextPageResponse msg = prepareNextPage(
+ nodeRess,
+ node,
+ qryResults,
+ req.query(),
+ req.segmentId(),
+ req.pageSize(),
+ dataPageScanEnabled);
+
+ if (msg != null)
+ sendNextPage(node, msg);
}
finally {
- res.unlock();
+ try {
+ res.unlockTables();
+ }
+ finally {
+ res.unlock();
+ }
}
}
- }
- catch (Exception e) {
- QueryRetryException retryEx = X.cause(e, QueryRetryException.class);
+ catch (Exception e) {
+ QueryRetryException retryEx = X.cause(e, QueryRetryException.class);
- if (retryEx != null)
- sendError(node, reqId, retryEx);
- else {
- SQLException sqlEx = X.cause(e, SQLException.class);
+ if (retryEx != null)
+ sendError(node, reqId, retryEx);
+ else {
+ SQLException sqlEx = X.cause(e, SQLException.class);
- if (sqlEx != null && sqlEx.getErrorCode() == ErrorCode.STATEMENT_WAS_CANCELED)
- sendQueryCancel(node, reqId);
- else
- sendError(node, reqId, e);
- }
+ if (sqlEx != null && sqlEx.getErrorCode() == ErrorCode.STATEMENT_WAS_CANCELED)
+ sendQueryCancel(node, reqId);
+ else
+ sendError(node, reqId, e);
+ }
- qryResults.cancel();
+ qryResults.cancel();
+ }
}
}
}
@@ -826,41 +865,45 @@ public class GridMapQueryExecutor {
int segmentId,
int pageSize,
Boolean dataPageScanEnabled) throws IgniteCheckedException {
- MapQueryResult res = qr.result(qry);
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_PAGE_PREPARE, MTC.span()))) {
+ MapQueryResult res = qr.result(qry);
- assert res != null;
+ assert res != null;
- if (res.closed())
- return null;
+ if (res.closed())
+ return null;
- int page = res.page();
+ int page = res.page();
- List<Value[]> rows = new ArrayList<>(Math.min(64, pageSize));
+ List<Value[]> rows = new ArrayList<>(Math.min(64, pageSize));
- boolean last = res.fetchNextPage(rows, pageSize, dataPageScanEnabled);
+ boolean last = res.fetchNextPage(rows, pageSize, dataPageScanEnabled);
- if (last) {
- qr.closeResult(qry);
+ if (last) {
+ qr.closeResult(qry);
- if (qr.isAllClosed()) {
- nodeRess.remove(qr.queryRequestId(), segmentId, qr);
+ if (qr.isAllClosed()) {
+ nodeRess.remove(qr.queryRequestId(), segmentId, qr);
- // Clear context, release reservations
- if (qr.isLazy())
- qr.releaseQueryContext();
+ // Clear context, release reservations
+ if (qr.isLazy())
+ qr.releaseQueryContext();
+ }
}
- }
- boolean loc = node.isLocal();
+ boolean loc = node.isLocal();
+
+ GridQueryNextPageResponse msg = new GridQueryNextPageResponse(qr.queryRequestId(), segmentId, qry, page,
+ page == 0 ? res.rowCount() : -1,
+ res.columnCount(),
+ loc ? null : toMessages(rows, new ArrayList<>(res.columnCount()), res.columnCount()),
+ loc ? rows : null,
+ last);
- GridQueryNextPageResponse msg = new GridQueryNextPageResponse(qr.queryRequestId(), segmentId, qry, page,
- page == 0 ? res.rowCount() : -1,
- res.columnCount(),
- loc ? null : toMessages(rows, new ArrayList<>(res.columnCount()), res.columnCount()),
- loc ? rows : null,
- last);
+ MTC.span().addTag(SQL_PAGE_ROWS, () -> String.valueOf(rows.size()));
- return msg;
+ return msg;
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
index 8f98592..cd4910d 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/GridReduceQueryExecutor.java
@@ -74,6 +74,8 @@ import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQuery
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlRequest;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2DmlResponse;
import org.apache.ignite.internal.processors.query.h2.twostep.msg.GridH2QueryRequest;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.transactions.IgniteTxAlreadyCompletedCheckedException;
import org.apache.ignite.internal.util.typedef.C2;
import org.apache.ignite.internal.util.typedef.CIX2;
@@ -99,6 +101,10 @@ import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.checkAc
import static org.apache.ignite.internal.processors.cache.mvcc.MvccUtils.tx;
import static org.apache.ignite.internal.processors.cache.query.GridCacheSqlQuery.EMPTY_PARAMS;
import static org.apache.ignite.internal.processors.query.h2.sql.GridSqlQuerySplitter.mergeTableIdentifier;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.ERROR;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_DML_QRY_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_FAIL_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_RESP;
/**
* Reduce query executor.
@@ -202,9 +208,11 @@ public class GridReduceQueryExecutor {
* @param msg Message.
*/
public void onFail(ClusterNode node, GridQueryFailResponse msg) {
- ReduceQueryRun r = runs.get(msg.queryRequestId());
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_FAIL_RESP, MTC.span()))) {
+ ReduceQueryRun r = runs.get(msg.queryRequestId());
- fail(r, node.id(), msg.error(), msg.failCode());
+ fail(r, node.id(), msg.error(), msg.failCode());
+ }
}
/**
@@ -235,62 +243,66 @@ public class GridReduceQueryExecutor {
* @param msg Message.
*/
public void onNextPage(final ClusterNode node, final GridQueryNextPageResponse msg) {
- final long qryReqId = msg.queryRequestId();
- final int qry = msg.query();
- final int seg = msg.segmentId();
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_PAGE_RESP, MTC.span()))) {
+ final long qryReqId = msg.queryRequestId();
+ final int qry = msg.query();
+ final int seg = msg.segmentId();
- final ReduceQueryRun r = runs.get(qryReqId);
+ final ReduceQueryRun r = runs.get(qryReqId);
- if (r == null) // Already finished with error or canceled.
- return;
+ if (r == null) // Already finished with error or canceled.
+ return;
- final int pageSize = r.pageSize();
+ final int pageSize = r.pageSize();
- Reducer idx = r.reducers().get(msg.query());
+ Reducer idx = r.reducers().get(msg.query());
- ReduceResultPage page;
+ ReduceResultPage page;
- try {
- page = new ReduceResultPage(ctx, node.id(), msg) {
- @Override public void fetchNextPage() {
- if (r.hasErrorOrRetry()) {
- if (r.exception() != null)
- throw r.exception();
+ try {
+ page = new ReduceResultPage(ctx, node.id(), msg) {
+ @Override public void fetchNextPage() {
+ if (r.hasErrorOrRetry()) {
+ if (r.exception() != null)
+ throw r.exception();
- assert r.retryCause() != null;
+ assert r.retryCause() != null;
- throw new CacheException(r.retryCause());
- }
+ throw new CacheException(r.retryCause());
+ }
- try {
- GridQueryNextPageRequest msg0 = new GridQueryNextPageRequest(qryReqId, qry, seg, pageSize,
- (byte)GridH2QueryRequest.setDataPageScanEnabled(0, r.isDataPageScanEnabled()));
+ try {
+ GridQueryNextPageRequest msg0 = new GridQueryNextPageRequest(qryReqId, qry, seg, pageSize,
+ (byte)GridH2QueryRequest.setDataPageScanEnabled(0, r.isDataPageScanEnabled()));
- if (node.isLocal())
- h2.mapQueryExecutor().onNextPageRequest(node, msg0);
- else
- ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg0, GridIoPolicy.QUERY_POOL);
- }
- catch (IgniteCheckedException e) {
- throw new CacheException("Failed to fetch data from node: " + node.id(), e);
+ if (node.isLocal())
+ h2.mapQueryExecutor().onNextPageRequest(node, msg0);
+ else
+ ctx.io().sendToGridTopic(node, GridTopic.TOPIC_QUERY, msg0, GridIoPolicy.QUERY_POOL);
+ }
+ catch (IgniteCheckedException e) {
+ throw new CacheException("Failed to fetch data from node: " + node.id(), e);
+ }
}
- }
- };
- }
- catch (Exception e) {
- U.error(log, "Error in message.", e);
+ };
+ }
+ catch (Exception e) {
+ U.error(log, "Error in message.", e);
- fail(r, node.id(), "Error in message.", GridQueryFailResponse.GENERAL_ERROR);
+ MTC.span().addTag(ERROR, e::getMessage);
- return;
- }
+ fail(r, node.id(), "Error in message.", GridQueryFailResponse.GENERAL_ERROR);
- idx.addPage(page);
+ return;
+ }
- if (msg.retry() != null)
- r.setStateOnRetry(node.id(), msg.retry(), msg.retryCause());
- else if (msg.page() == 0) // Count down only on each first page received.
- r.onFirstPage();
+ idx.addPage(page);
+
+ if (msg.retry() != null)
+ r.setStateOnRetry(node.id(), msg.retry(), msg.retryCause());
+ else if (msg.page() == 0) // Count down only on each first page received.
+ r.onFirstPage();
+ }
}
/**
@@ -480,7 +492,8 @@ public class GridReduceQueryExecutor {
r,
qryReqId,
qry.distributedJoins(),
- mvccTracker);
+ mvccTracker,
+ ctx.tracing());
release = false;
@@ -520,7 +533,16 @@ public class GridReduceQueryExecutor {
qryInfo
);
- resIter = new H2FieldsIterator(res, mvccTracker, conn, r.pageSize(), log, h2, qryInfo);
+ resIter = new H2FieldsIterator(
+ res,
+ mvccTracker,
+ conn,
+ r.pageSize(),
+ log,
+ h2,
+ qryInfo,
+ ctx.tracing()
+ );
conn = null;
@@ -922,7 +944,7 @@ public class GridReduceQueryExecutor {
* @param msg Message.
*/
public void onDmlResponse(final ClusterNode node, GridH2DmlResponse msg) {
- try {
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_DML_QRY_RESP, MTC.span()))) {
long reqId = msg.requestId();
DmlDistributedUpdateRun r = updRuns.get(reqId);
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java
index 4177594..d7c9dba 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/PartitionReservationManager.java
@@ -37,6 +37,8 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.Par
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionsReservation;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import org.apache.ignite.internal.util.lang.GridPlainRunnable;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.F;
@@ -45,6 +47,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion.NONE;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PARTITIONS_RESERVE;
/**
* Class responsible for partition reservation for queries executed on local node. Prevents partitions from being
@@ -104,110 +107,130 @@ public class PartitionReservationManager implements PartitionsExchangeAware {
UUID nodeId,
long reqId
) throws IgniteCheckedException {
- assert reqTopVer != null;
+ try (TraceSurroundings ignored = MTC.support(ctx.tracing().create(SQL_PARTITIONS_RESERVE, MTC.span()))) {
+ assert reqTopVer != null;
- AffinityTopologyVersion topVer = ctx.cache().context().exchange().lastAffinityChangedTopologyVersion(reqTopVer);
+ AffinityTopologyVersion topVer = ctx.cache().context().exchange().lastAffinityChangedTopologyVersion(reqTopVer);
- if (F.isEmpty(cacheIds))
- return new PartitionReservation(Collections.emptyList());
+ if (F.isEmpty(cacheIds))
+ return new PartitionReservation(Collections.emptyList());
- Collection<Integer> partIds;
+ Collection<Integer> partIds;
- if (explicitParts == null)
- partIds = null;
- else if (explicitParts.length == 0)
- partIds = Collections.emptyList();
- else {
- partIds = new ArrayList<>(explicitParts.length);
+ if (explicitParts == null)
+ partIds = null;
+ else if (explicitParts.length == 0)
+ partIds = Collections.emptyList();
+ else {
+ partIds = new ArrayList<>(explicitParts.length);
- for (int explicitPart : explicitParts)
- partIds.add(explicitPart);
- }
+ for (int explicitPart : explicitParts)
+ partIds.add(explicitPart);
+ }
- List<GridReservable> reserved = new ArrayList<>();
+ List<GridReservable> reserved = new ArrayList<>();
- for (int i = 0; i < cacheIds.size(); i++) {
- GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(cacheIds.get(i));
+ for (int i = 0; i < cacheIds.size(); i++) {
+ GridCacheContext<?, ?> cctx = ctx.cache().context().cacheContext(cacheIds.get(i));
- // Cache was not found, probably was not deployed yet.
- if (cctx == null) {
- return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions for query (cache is not " +
- "found on local node) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s]",
- ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i)));
- }
+ // Cache was not found, probably was not deployed yet.
+ if (cctx == null) {
+ return new PartitionReservation(reserved,
+ String.format("Failed to reserve partitions for query (cache is not " +
+ "found on local node) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s]",
+ ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i)));
+ }
- if (cctx.isLocal() || !cctx.rebalanceEnabled())
- continue;
+ if (cctx.isLocal() || !cctx.rebalanceEnabled())
+ continue;
- // For replicated cache topology version does not make sense.
- final PartitionReservationKey grpKey = new PartitionReservationKey(cctx.name(), cctx.isReplicated() ? null : topVer);
+ // For replicated cache topology version does not make sense.
+ final PartitionReservationKey grpKey = new PartitionReservationKey(cctx.name(), cctx.isReplicated() ? null : topVer);
- GridReservable r = reservations.get(grpKey);
+ GridReservable r = reservations.get(grpKey);
- if (explicitParts == null && r != null) { // Try to reserve group partition if any and no explicits.
- if (r != REPLICATED_RESERVABLE) {
- if (!r.reserve())
- return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions for query (group " +
- "reservation failed) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s]", ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i), cctx.name()));
+ if (explicitParts == null && r != null) { // Try to reserve group partition if any and no explicits.
+ if (r != REPLICATED_RESERVABLE) {
+ if (!r.reserve())
+ return new PartitionReservation(reserved,
+ String.format("Failed to reserve partitions for query (group " +
+ "reservation failed) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s]", ctx.localNodeId(), nodeId, reqId, topVer, cacheIds.get(i), cctx.name()));
- reserved.add(r);
+ reserved.add(r);
+ }
}
- }
- else { // Try to reserve partitions one by one.
- int partsCnt = cctx.affinity().partitions();
-
- if (cctx.isReplicated()) { // Check all the partitions are in owning state for replicated cache.
- if (r == null) { // Check only once.
- for (int p = 0; p < partsCnt; p++) {
- GridDhtLocalPartition part = partition(cctx, p);
-
- // We don't need to reserve partitions because they will not be evicted in replicated caches.
- GridDhtPartitionState partState = part != null ? part.state() : null;
+ else { // Try to reserve partitions one by one.
+ int partsCnt = cctx.affinity().partitions();
+
+ if (cctx.isReplicated()) { // Check all the partitions are in owning state for replicated cache.
+ if (r == null) { // Check only once.
+ for (int p = 0; p < partsCnt; p++) {
+ GridDhtLocalPartition part = partition(cctx, p);
+
+ // We don't need to reserve partitions because they will not be evicted in replicated caches.
+ GridDhtPartitionState partState = part != null ? part.state() : null;
+
+ if (partState != OWNING)
+ return new PartitionReservation(reserved,
+ String.format("Failed to reserve partitions for " +
+ "query (partition of REPLICATED cache is not in OWNING state) [" +
+ "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s, part=%s, partFound=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ p,
+ (part != null),
+ partState
+ ));
+ }
- if (partState != OWNING)
- return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions for " +
- "query (partition of REPLICATED cache is not in OWNING state) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s, part=%s, partFound=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- p,
- (part != null),
- partState
- ));
+ // Mark that we checked this replicated cache.
+ reservations.putIfAbsent(grpKey, REPLICATED_RESERVABLE);
}
-
- // Mark that we checked this replicated cache.
- reservations.putIfAbsent(grpKey, REPLICATED_RESERVABLE);
}
- }
- else { // Reserve primary partitions for partitioned cache (if no explicit given).
- if (explicitParts == null)
- partIds = cctx.affinity().primaryPartitions(ctx.localNodeId(), topVer);
+ else { // Reserve primary partitions for partitioned cache (if no explicit given).
+ if (explicitParts == null)
+ partIds = cctx.affinity().primaryPartitions(ctx.localNodeId(), topVer);
+
+ int reservedCnt = 0;
- int reservedCnt = 0;
+ for (int partId : partIds) {
+ GridDhtLocalPartition part = partition(cctx, partId);
- for (int partId : partIds) {
- GridDhtLocalPartition part = partition(cctx, partId);
+ GridDhtPartitionState partState = part != null ? part.state() : null;
- GridDhtPartitionState partState = part != null ? part.state() : null;
+ if (partState != OWNING) {
+ if (partState == LOST)
+ failQueryOnLostData(cctx, part);
+ else {
+ return new PartitionReservation(reserved,
+ String.format("Failed to reserve partitions " +
+ "for query (partition of PARTITIONED cache is not found or not in OWNING " +
+ "state) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ "cacheName=%s, part=%s, partFound=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ partId,
+ (part != null),
+ partState
+ ));
+ }
+ }
- if (partState != OWNING) {
- if (partState == LOST)
- failQueryOnLostData(cctx, part);
- else {
+ if (!part.reserve()) {
return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions " +
- "for query (partition of PARTITIONED cache is not found or not in OWNING " +
- "state) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
+ String.format("Failed to reserve partitions for query " +
+ "(partition of PARTITIONED cache cannot be reserved) [" +
+ "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
"cacheName=%s, part=%s, partFound=%s, partState=%s]",
ctx.localNodeId(),
nodeId,
@@ -216,79 +239,61 @@ public class PartitionReservationManager implements PartitionsExchangeAware {
cacheIds.get(i),
cctx.name(),
partId,
- (part != null),
+ true,
partState
));
}
- }
-
- if (!part.reserve()) {
- return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions for query " +
- "(partition of PARTITIONED cache cannot be reserved) [" +
- "localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, cacheId=%s, " +
- "cacheName=%s, part=%s, partFound=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- partId,
- true,
- partState
- ));
- }
- reserved.add(part);
-
- reservedCnt++;
-
- // Double check that we are still in owning state and partition contents are not cleared.
- partState = part.state();
-
- if (partState != OWNING) {
- if (partState == LOST)
- failQueryOnLostData(cctx, part);
- else {
- return new PartitionReservation(reserved,
- String.format("Failed to reserve partitions for " +
- "query (partition of PARTITIONED cache is not in OWNING state after " +
- "reservation) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, " +
- "cacheId=%s, cacheName=%s, part=%s, partState=%s]",
- ctx.localNodeId(),
- nodeId,
- reqId,
- topVer,
- cacheIds.get(i),
- cctx.name(),
- partId,
- partState
- ));
+ reserved.add(part);
+
+ reservedCnt++;
+
+ // Double check that we are still in owning state and partition contents are not cleared.
+ partState = part.state();
+
+ if (partState != OWNING) {
+ if (partState == LOST)
+ failQueryOnLostData(cctx, part);
+ else {
+ return new PartitionReservation(reserved,
+ String.format("Failed to reserve partitions for " +
+ "query (partition of PARTITIONED cache is not in OWNING state after " +
+ "reservation) [localNodeId=%s, rmtNodeId=%s, reqId=%s, affTopVer=%s, " +
+ "cacheId=%s, cacheName=%s, part=%s, partState=%s]",
+ ctx.localNodeId(),
+ nodeId,
+ reqId,
+ topVer,
+ cacheIds.get(i),
+ cctx.name(),
+ partId,
+ partState
+ ));
+ }
}
}
- }
- if (explicitParts == null && reservedCnt > 0) {
- // We reserved all the primary partitions for cache, attempt to add group reservation.
- GridDhtPartitionsReservation grp = new GridDhtPartitionsReservation(topVer, cctx, "SQL");
+ if (explicitParts == null && reservedCnt > 0) {
+ // We reserved all the primary partitions for cache, attempt to add group reservation.
+ GridDhtPartitionsReservation grp = new GridDhtPartitionsReservation(topVer, cctx, "SQL");
- if (grp.register(reserved.subList(reserved.size() - reservedCnt, reserved.size()))) {
- if (reservations.putIfAbsent(grpKey, grp) != null)
- throw new IllegalStateException("Reservation already exists.");
+ if (grp.register(reserved.subList(reserved.size() - reservedCnt, reserved.size()))) {
+ if (reservations.putIfAbsent(grpKey, grp) != null)
+ throw new IllegalStateException("Reservation already exists.");
- grp.onPublish(new CI1<GridDhtPartitionsReservation>() {
- @Override public void apply(GridDhtPartitionsReservation r) {
- reservations.remove(grpKey, r);
- }
- });
+ grp.onPublish(new CI1<GridDhtPartitionsReservation>() {
+ @Override public void apply(GridDhtPartitionsReservation r) {
+ reservations.remove(grpKey, r);
+ }
+ });
+ }
}
}
}
}
- }
- return new PartitionReservation(reserved);
+ return new PartitionReservation(reserved);
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceIndexIterator.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceIndexIterator.java
index 8a035a6..20a9944 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceIndexIterator.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/ReduceIndexIterator.java
@@ -24,10 +24,15 @@ import java.util.List;
import java.util.NoSuchElementException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.processors.cache.mvcc.MvccQueryTracker;
+import org.apache.ignite.internal.processors.tracing.MTC;
+import org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
+import org.apache.ignite.internal.processors.tracing.Tracing;
import org.h2.index.Cursor;
import org.h2.result.Row;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_CLOSE;
+
/**
* Iterator that transparently and sequentially traverses a bunch of {@link AbstractReduceIndexAdapter} objects.
*/
@@ -62,6 +67,9 @@ public class ReduceIndexIterator implements Iterator<List<?>>, AutoCloseable {
/** */
private MvccQueryTracker mvccTracker;
+ /** Tracing processor. */
+ private final Tracing tracing;
+
/**
* Constructor.
*
@@ -70,13 +78,15 @@ public class ReduceIndexIterator implements Iterator<List<?>>, AutoCloseable {
* @param run Query run.
* @param qryReqId Query request ID.
* @param distributedJoins Distributed joins.
+ * @param tracing Tracing processor.
*/
public ReduceIndexIterator(GridReduceQueryExecutor rdcExec,
Collection<ClusterNode> nodes,
ReduceQueryRun run,
long qryReqId,
boolean distributedJoins,
- @Nullable MvccQueryTracker mvccTracker
+ @Nullable MvccQueryTracker mvccTracker,
+ Tracing tracing
) {
this.rdcExec = rdcExec;
this.nodes = nodes;
@@ -84,6 +94,7 @@ public class ReduceIndexIterator implements Iterator<List<?>>, AutoCloseable {
this.qryReqId = qryReqId;
this.distributedJoins = distributedJoins;
this.mvccTracker = mvccTracker;
+ this.tracing = tracing;
rdcIter = run.reducers().iterator();
@@ -114,7 +125,9 @@ public class ReduceIndexIterator implements Iterator<List<?>>, AutoCloseable {
/** {@inheritDoc} */
@Override public void close() throws Exception {
- releaseIfNeeded();
+ try (TraceSurroundings ignored = MTC.support(tracing.create(SQL_ITER_CLOSE, MTC.span()))) {
+ releaseIfNeeded();
+ }
}
/**
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/msg/GridH2RowRange.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/msg/GridH2RowRange.java
index 3a7e6ac..2c4820c 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/msg/GridH2RowRange.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/msg/GridH2RowRange.java
@@ -174,8 +174,15 @@ public class GridH2RowRange implements Message {
// No-op.
}
+ /**
+ * @return Number of rows.
+ */
+ public int rowsSize() {
+ return rows == null ? 0 : rows.size();
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
- return S.toString(GridH2RowRange.class, this, "rowsSize", rows != null ? rows.size() : 0);
+ return S.toString(GridH2RowRange.class, this, "rowsSize", rowsSize());
}
}
diff --git a/modules/opencensus/pom.xml b/modules/opencensus/pom.xml
index c9d8797..648c2bc 100644
--- a/modules/opencensus/pom.xml
+++ b/modules/opencensus/pom.xml
@@ -108,5 +108,12 @@
<version>${commons.io.version}</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.ignite</groupId>
+ <artifactId>ignite-indexing</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/AbstractTracingTest.java b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/AbstractTracingTest.java
index c7a8a26..007e821 100644
--- a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/AbstractTracingTest.java
+++ b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/AbstractTracingTest.java
@@ -109,6 +109,10 @@ public abstract class AbstractTracingTest extends GridCommonAbstractTest {
DFLT_CONFIG_MAP.put(
new TracingConfigurationCoordinates.Builder(Scope.DISCOVERY).build(),
TracingConfigurationManager.DEFAULT_DISCOVERY_CONFIGURATION);
+
+ DFLT_CONFIG_MAP.put(
+ new TracingConfigurationCoordinates.Builder(Scope.SQL).build(),
+ TracingConfigurationManager.DEFAULT_SQL_CONFIGURATION);
}
/** Test trace exporter handler. */
diff --git a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/IgniteOpenCensusSuite.java b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/IgniteOpenCensusSuite.java
index ab5e1d8..1510bca 100644
--- a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/IgniteOpenCensusSuite.java
+++ b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/IgniteOpenCensusSuite.java
@@ -36,7 +36,9 @@ import org.junit.runners.Suite;
OpenCensusTracingConfigurationGetAllTest.class,
OpenCensusTracingConfigurationResetTest.class,
OpenCensusTracingConfigurationResetAllTest.class,
- OpenCensusDiscoveryTracingTest.class
+ OpenCensusDiscoveryTracingTest.class,
+ OpenCensusSqlJdbcTracingTest.class,
+ OpenCensusSqlNativeTracingTest.class
})
public class IgniteOpenCensusSuite {
}
diff --git a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlJdbcTracingTest.java b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlJdbcTracingTest.java
new file mode 100644
index 0000000..fe9aa1a
--- /dev/null
+++ b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlJdbcTracingTest.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.monitoring.opencensus;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Objects;
+import io.opencensus.trace.SpanId;
+import org.apache.ignite.client.Config;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.spi.tracing.TracingConfigurationCoordinates;
+import org.apache.ignite.spi.tracing.TracingConfigurationParameters;
+import org.junit.Test;
+
+import static java.sql.DriverManager.getConnection;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.internal.processors.query.QueryUtils.DFLT_SCHEMA;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_PAGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_BATCH_PROCESS;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CMD_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_CLOSE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_CLOSE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_FETCH;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_PARSE;
+import static org.apache.ignite.internal.util.IgniteUtils.resolveIgnitePath;
+import static org.apache.ignite.spi.tracing.Scope.SQL;
+import static org.apache.ignite.spi.tracing.TracingConfigurationParameters.SAMPLING_RATE_ALWAYS;
+import static org.apache.ignite.spi.tracing.TracingConfigurationParameters.SAMPLING_RATE_NEVER;
+
+/**
+ * Tests tracing of SQL quries execution via JDBC.
+ */
+public class OpenCensusSqlJdbcTracingTest extends OpenCensusSqlNativeTracingTest {
+ /** JDBC URL prefix. */
+ private static final String JDBC_URL_PREFIX = "jdbc:ignite:thin://";
+
+ /** CSV file for bulk-load testing. */
+ private static final String BULKLOAD_FILE = Objects.requireNonNull(resolveIgnitePath(
+ "/modules/clients/src/test/resources/bulkload2.csv")).getAbsolutePath();
+
+ /** Number of bulk-load entries. */
+ private static final int BULKLOAD_ENTRIES = 2;
+
+ /**
+ * Tests tracing of local SQL SELECT query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectLocal() throws Exception {
+ String orgTable = createTableAndPopulate(Organization.class, REPLICATED, 1);
+
+ SpanId rootSpan = executeAndCheckRootSpan("SELECT orgVal FROM " + orgTable, TEST_SCHEMA, false, false, true);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+ checkChildSpan(SQL_CURSOR_OPEN, rootSpan);
+ checkChildSpan(SQL_ITER_OPEN, rootSpan);
+
+ SpanId iterSpan = checkChildSpan(SQL_ITER_OPEN, rootSpan);
+
+ checkChildSpan(SQL_QRY_EXECUTE, iterSpan);
+
+ int fetchedRows = findChildSpans(SQL_PAGE_FETCH, rootSpan).stream()
+ .mapToInt(span -> getAttribute(span, SQL_PAGE_ROWS))
+ .sum();
+
+ assertEquals(TEST_TABLE_POPULATION, fetchedRows);
+
+ checkChildSpan(SQL_ITER_CLOSE, rootSpan);
+ assertFalse(findChildSpans(SQL_CURSOR_CLOSE, rootSpan).isEmpty());
+ }
+
+ /**
+ * Test SQL bulk load query tracing.
+ */
+ @Test
+ public void testCopy() throws Exception {
+ ignite(0).tracingConfiguration().set(
+ new TracingConfigurationCoordinates.Builder(SQL).build(),
+ new TracingConfigurationParameters.Builder()
+ .withSamplingRate(SAMPLING_RATE_NEVER).build());
+
+ String table = "test_table";
+
+ executeQuery(
+ "CREATE TABLE " + table + "(id LONG PRIMARY KEY, first_name VARCHAR, last_name VARCHAR, age LONG)",
+ DFLT_SCHEMA, false, false, null);
+
+ ignite(0).tracingConfiguration().set(
+ new TracingConfigurationCoordinates.Builder(SQL).build(),
+ new TracingConfigurationParameters.Builder()
+ .withSamplingRate(SAMPLING_RATE_ALWAYS).build());
+
+ SpanId rootSpan = executeAndCheckRootSpan(
+ "COPY FROM '" + BULKLOAD_FILE + "' INTO " + table + "(id, age, first_name, last_name) FORMAT csv",
+ DFLT_SCHEMA, false, false, false);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+ checkChildSpan(SQL_CMD_QRY_EXECUTE, rootSpan);
+ checkSpan(SQL_BATCH_PROCESS, rootSpan, BULKLOAD_ENTRIES, null);
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("StatementWithEmptyBody")
+ @Override protected void executeQuery(
+ String sql,
+ String schema,
+ boolean skipReduceOnUpdate,
+ boolean distributedJoins,
+ Boolean isQry
+ ) {
+ String url = JDBC_URL_PREFIX + Config.SERVER + '/' + schema +
+ "?skipReducerOnUpdate=" + skipReduceOnUpdate + "&distributedJoins=" + distributedJoins;
+
+ try (Connection conn = getConnection(url)) {
+ PreparedStatement stmt = conn.prepareStatement(sql);
+
+ stmt.setFetchSize(PAGE_SIZE);
+
+ if (isQry == null)
+ stmt.execute();
+ else if (!isQry)
+ stmt.executeUpdate();
+ else {
+ stmt.executeQuery();
+
+ try (ResultSet resultSet = stmt.getResultSet()) {
+ while (resultSet.next()) {
+ // No-op.
+ }
+ }
+ }
+ }
+ catch (SQLException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteEx reducer() {
+ return ignite(0);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected int mapNodesCount() {
+ return GRID_CNT - 1;
+ }
+}
diff --git a/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlNativeTracingTest.java b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlNativeTracingTest.java
new file mode 100644
index 0000000..d48f91c
--- /dev/null
+++ b/modules/opencensus/src/test/java/org/apache/ignite/internal/processors/monitoring/opencensus/OpenCensusSqlNativeTracingTest.java
@@ -0,0 +1,646 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.monitoring.opencensus;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import com.google.common.collect.ImmutableMap;
+import io.opencensus.trace.SpanId;
+import io.opencensus.trace.Tracing;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.query.SqlFieldsQuery;
+import org.apache.ignite.cache.query.annotations.QuerySqlField;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.processors.cache.query.SqlFieldsQueryEx;
+import org.apache.ignite.internal.processors.query.h2.twostep.messages.GridQueryNextPageRequest;
+import org.apache.ignite.internal.processors.tracing.SpanType;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.tracing.TracingConfigurationCoordinates;
+import org.apache.ignite.spi.tracing.TracingConfigurationParameters;
+import org.apache.ignite.spi.tracing.TracingSpi;
+import org.apache.ignite.spi.tracing.opencensus.OpenCensusTracingSpi;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+import static java.lang.Integer.parseInt;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi;
+import static org.apache.ignite.internal.processors.query.QueryUtils.DFLT_SCHEMA;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.CONSISTENT_ID;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.NAME;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.NODE;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.NODE_ID;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_CACHE_UPDATES;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_IDX_RANGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_PAGE_ROWS;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_QRY_TEXT;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.SQL_SCHEMA;
+import static org.apache.ignite.internal.processors.tracing.SpanTags.tag;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CACHE_UPDATE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CMD_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_CANCEL;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_CLOSE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_CURSOR_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_DML_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_DML_QRY_EXEC_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_DML_QRY_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_FAIL_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_IDX_RANGE_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_IDX_RANGE_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_CLOSE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_ITER_OPEN;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_NEXT_PAGE_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_FETCH;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_PREPARE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_RESP;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PAGE_WAIT;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_PARTITIONS_RESERVE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_CANCEL_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_EXECUTE;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_EXEC_REQ;
+import static org.apache.ignite.internal.processors.tracing.SpanType.SQL_QRY_PARSE;
+import static org.apache.ignite.spi.tracing.Scope.SQL;
+import static org.apache.ignite.spi.tracing.TracingConfigurationParameters.SAMPLING_RATE_ALWAYS;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+
+/**
+ * Tests tracing of SQL queries based on {@link OpenCensusTracingSpi}.
+ */
+public class OpenCensusSqlNativeTracingTest extends AbstractTracingTest {
+ /** Number of entries in test table. */
+ protected static final int TEST_TABLE_POPULATION = 100;
+
+ /** Page size for queries. */
+ protected static final int PAGE_SIZE = 20;
+
+ /** Test schema name. */
+ protected static final String TEST_SCHEMA = "TEST_SCHEMA";
+
+ /** Key counter. */
+ private final AtomicInteger keyCntr = new AtomicInteger();
+
+ /** {@inheritDoc} */
+ @Override protected TracingSpi<?> getTracingSpi() {
+ return new OpenCensusTracingSpi();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setCommunicationSpi(new TestRecordingCommunicationSpi());
+ }
+
+ /** {@inheritDoc} */
+ @Override public void before() throws Exception {
+ super.before();
+
+ IgniteEx cli = startClientGrid(GRID_CNT);
+
+ cli.tracingConfiguration().set(
+ new TracingConfigurationCoordinates.Builder(SQL).build(),
+ new TracingConfigurationParameters.Builder()
+ .withSamplingRate(SAMPLING_RATE_ALWAYS).build());
+ }
+
+ /**
+ * Tests tracing of UPDATE query with skipped reduce phase.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testUpdateWithReducerSkipped() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ SpanId rootSpan = executeAndCheckRootSpan(
+ "UPDATE " + prsnTable + " SET prsnVal = (prsnVal + 1)", TEST_SCHEMA, true, false, false);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+
+ SpanId dmlExecSpan = checkChildSpan(SQL_DML_QRY_EXECUTE, rootSpan);
+
+ List<SpanId> execReqSpans = checkSpan(SQL_DML_QRY_EXEC_REQ, dmlExecSpan, GRID_CNT, null);
+
+ int fetchedRows = 0;
+
+ int cacheUpdates = 0;
+
+ assertEquals(GRID_CNT, execReqSpans.size());
+
+ for (int i = 0; i < GRID_CNT; i++) {
+ SpanId execReqSpan = execReqSpans.get(i);
+
+ checkChildSpan(SQL_PARTITIONS_RESERVE, execReqSpan);
+ checkSpan(SQL_QRY_PARSE, execReqSpan, 2, null);
+
+ SpanId iterSpan = checkChildSpan(SQL_ITER_OPEN, execReqSpan);
+
+ checkChildSpan(SQL_QRY_EXECUTE, iterSpan);
+
+ fetchedRows += findChildSpans(SQL_PAGE_FETCH, execReqSpan).stream()
+ .mapToInt(span -> getAttribute(span, SQL_PAGE_ROWS))
+ .sum();
+
+ List<SpanId> cacheUpdateSpans = findChildSpans(SQL_CACHE_UPDATE, execReqSpan);
+
+ cacheUpdates += cacheUpdateSpans.stream()
+ .mapToInt(span -> getAttribute(span, SQL_CACHE_UPDATES))
+ .sum();
+
+ checkChildSpan(SQL_ITER_CLOSE, execReqSpan);
+ checkChildSpan(SQL_DML_QRY_RESP, execReqSpan);
+ }
+
+ assertEquals(TEST_TABLE_POPULATION, fetchedRows);
+ assertEquals(TEST_TABLE_POPULATION, cacheUpdates);
+ }
+
+ /**
+ * Tests tracing of multiple MERGE query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testMultipleMerge() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("MERGE INTO " + prsnTable + "(_key, prsnVal) SELECT _key, 0 FROM " + prsnTable,
+ true, TEST_TABLE_POPULATION);
+ }
+
+ /**
+ * Tests tracing of single MERGE query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSingleMerge() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("MERGE INTO " + prsnTable + "(_key, prsnId, prsnVal) VALUES (" + keyCntr.get() + ", 0, 0)",
+ false, 1);
+ }
+
+ /**
+ * Tests tracing of UPDATE query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testUpdate() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("UPDATE " + prsnTable + " SET prsnVal = (prsnVal + 1)",
+ true, TEST_TABLE_POPULATION);
+ }
+
+ /**
+ * Tests tracing of DELETE query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testDelete() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("DELETE FROM " + prsnTable, true, TEST_TABLE_POPULATION);
+ }
+
+ /**
+ * Tests tracing of multiple INSERT query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testMultipleInsert() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("INSERT INTO " + prsnTable + "(_key, prsnId, prsnVal) VALUES" +
+ " (" + keyCntr.incrementAndGet() + ", 0, 0)," +
+ " (" + keyCntr.incrementAndGet() + ", 1, 1)", false, 2);
+ }
+
+ /**
+ * Tests tracing of single INSERT query.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSingleInsert() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ checkDmlQuerySpans("INSERT INTO " + prsnTable + "(_key, prsnId, prsnVal) VALUES" +
+ " (" + keyCntr.incrementAndGet() + ", 0, 0)", false, 1);
+ }
+
+ /**
+ * Tests tracing of distributed join query which includes all communications between reducer and mapped nodes and
+ * index range requests.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testDistributedJoin() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ String orgTable = createTableAndPopulate(Organization.class, PARTITIONED, 1);
+
+ SpanId rootSpan = executeAndCheckRootSpan(
+ "SELECT * FROM " + prsnTable + " AS p JOIN " + orgTable + " AS o ON o.orgId = p.prsnId",
+ TEST_SCHEMA, false, true, true);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+ checkChildSpan(SQL_CURSOR_OPEN, rootSpan);
+
+ SpanId iterSpan = checkChildSpan(SQL_ITER_OPEN, rootSpan);
+
+ List<SpanId> execReqSpans = checkSpan(SQL_QRY_EXEC_REQ, iterSpan, GRID_CNT, null);
+
+ int idxRangeReqRows = 0;
+
+ int preparedRows = 0;
+
+ int fetchedRows = 0;
+
+ for (int i = 0; i < GRID_CNT; i++) {
+ SpanId execReqSpan = execReqSpans.get(i);
+
+ checkChildSpan(SQL_PARTITIONS_RESERVE, execReqSpan);
+
+ SpanId execSpan = checkChildSpan(SQL_QRY_EXECUTE, execReqSpan);
+
+ List<SpanId> distrLookupReqSpans = findChildSpans(SQL_IDX_RANGE_REQ, execSpan);
+
+ for (SpanId span : distrLookupReqSpans) {
+ idxRangeReqRows += getAttribute(span, SQL_IDX_RANGE_ROWS);
+
+ checkChildSpan(SQL_IDX_RANGE_RESP, span);
+ }
+
+ preparedRows += getAttribute(
+ checkChildSpan(SQL_PAGE_PREPARE, execReqSpan), SQL_PAGE_ROWS);
+
+ checkChildSpan(SQL_PAGE_RESP, execReqSpan);
+ }
+
+ SpanId pageFetchSpan = checkChildSpan(SQL_PAGE_FETCH, iterSpan);
+
+ fetchedRows += getAttribute(pageFetchSpan, SQL_PAGE_ROWS);
+
+ checkChildSpan(SQL_PAGE_WAIT, pageFetchSpan);
+
+ SpanId nexPageSpan = checkChildSpan(SQL_NEXT_PAGE_REQ, pageFetchSpan);
+
+ preparedRows += getAttribute(
+ checkChildSpan(SQL_PAGE_PREPARE, nexPageSpan), SQL_PAGE_ROWS);
+
+ checkChildSpan(SQL_PAGE_RESP, nexPageSpan);
+
+ List<SpanId> pageFetchSpans = findChildSpans(SQL_PAGE_FETCH, rootSpan);
+
+ for (SpanId span : pageFetchSpans) {
+ fetchedRows += getAttribute(span, SQL_PAGE_ROWS);
+
+ checkChildSpan(SQL_PAGE_WAIT, span);
+
+ List<SpanId> nextPageSpans = findChildSpans(SQL_NEXT_PAGE_REQ, span);
+
+ if (!nextPageSpans.isEmpty()) {
+ assertEquals(1, nextPageSpans.size());
+
+ SpanId nextPageSpan = nextPageSpans.get(0);
+
+ preparedRows += getAttribute(
+ checkChildSpan(SQL_PAGE_PREPARE, nextPageSpan), SQL_PAGE_ROWS);
+
+ checkChildSpan(SQL_PAGE_RESP, nextPageSpan);
+ }
+ }
+
+ assertEquals(TEST_TABLE_POPULATION, fetchedRows);
+ assertEquals(TEST_TABLE_POPULATION, preparedRows);
+ assertEquals(TEST_TABLE_POPULATION, idxRangeReqRows);
+
+ checkSpan(SQL_QRY_CANCEL_REQ, rootSpan, mapNodesCount(), null);
+
+ assertFalse(findChildSpans(SQL_CURSOR_CLOSE, rootSpan).isEmpty());
+ }
+
+ /**
+ * Tests tracing of SELECT query with parallelism.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testSelectWithParallelism() throws Exception {
+ int qryParallelism = 2;
+
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, qryParallelism);
+
+ SpanId rootSpan = executeAndCheckRootSpan("SELECT * FROM " + prsnTable,
+ TEST_SCHEMA, false, false, true);
+
+ SpanId iterOpenSpan = checkChildSpan(SQL_ITER_OPEN, rootSpan);
+
+ List<SpanId> qryExecspans = findChildSpans(SQL_QRY_EXEC_REQ, iterOpenSpan);
+
+ assertEquals(GRID_CNT * qryParallelism, qryExecspans.size());
+ }
+
+ /**
+ * Tests tracing of the SQL query next page request failure.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ @SuppressWarnings("Convert2MethodRef")
+ public void testNextPageRequestFailure() throws Exception {
+ String prsnTable = createTableAndPopulate(Person.class, PARTITIONED, 1);
+
+ IgniteEx qryInitiator = reducer();
+
+ spi(qryInitiator).blockMessages((node, msg) -> msg instanceof GridQueryNextPageRequest);
+
+ IgniteInternalFuture<?> iterFut = runAsync(() ->
+ executeQuery("SELECT * FROM " + prsnTable, TEST_SCHEMA, false, false, true));
+
+ spi(qryInitiator).waitForBlocked(mapNodesCount());
+
+ qryInitiator.context().query().runningQueries(-1).iterator().next().cancel();
+
+ spi(qryInitiator).stopBlock();
+
+ GridTestUtils.assertThrowsWithCause(() -> iterFut.get(), IgniteCheckedException.class);
+
+ handler().flush();
+
+ checkDroppedSpans();
+
+ SpanId rootSpan = checkChildSpan(SQL_QRY, null);
+
+ SpanId cursorCancelSpan = checkChildSpan(SQL_CURSOR_CANCEL, rootSpan);
+
+ SpanId cursorCloseSpan = checkChildSpan(SQL_CURSOR_CLOSE, cursorCancelSpan);
+
+ SpanId iterCloseSpan = checkChildSpan(SQL_ITER_CLOSE, cursorCloseSpan);
+
+ checkSpan(SQL_QRY_CANCEL_REQ, iterCloseSpan, mapNodesCount(), null);
+
+ List<SpanId> pageFetchSpans = findChildSpans(SQL_PAGE_FETCH, rootSpan);
+
+ for (SpanId pageFetchSpan : pageFetchSpans) {
+ List<SpanId> nextPageReqSpans = findChildSpans(SQL_NEXT_PAGE_REQ, pageFetchSpan);
+
+ if (!nextPageReqSpans.isEmpty()) {
+ assertEquals(1, nextPageReqSpans.size());
+
+ checkChildSpan(SQL_FAIL_RESP, nextPageReqSpans.get(0));
+ }
+ }
+ }
+
+ /**
+ * Tests tracing of the CREATE TABLE command execution.
+ */
+ @Test
+ public void testCreateTable() throws Exception {
+ SpanId rootSpan = executeAndCheckRootSpan("CREATE TABLE test_table(id INT PRIMARY KEY, val VARCHAR)",
+ DFLT_SCHEMA, false, false, null);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+ checkChildSpan(SQL_CMD_QRY_EXECUTE, rootSpan);
+ }
+
+ /**
+ * Executes DML query and checks corresponding span tree.
+ *
+ * @param qry SQL query to execute.
+ * @param fetchRequired Whether query need to fetch data before cache update.
+ */
+ private void checkDmlQuerySpans(String qry, boolean fetchRequired, int expCacheUpdates) throws Exception {
+ SpanId rootSpan = executeAndCheckRootSpan(qry, TEST_SCHEMA, false,false,false);
+
+ checkChildSpan(SQL_QRY_PARSE, rootSpan);
+
+ SpanId dmlExecSpan = checkChildSpan(SQL_DML_QRY_EXECUTE, rootSpan);
+
+ if (fetchRequired) {
+ checkChildSpan(SQL_ITER_OPEN, dmlExecSpan);
+
+ int fetchedRows = findChildSpans(SQL_PAGE_FETCH, null).stream()
+ .mapToInt(span -> getAttribute(span, SQL_PAGE_ROWS))
+ .sum();
+
+ assertEquals(expCacheUpdates, fetchedRows);
+ }
+
+ int cacheUpdates = findChildSpans(SQL_CACHE_UPDATE, dmlExecSpan).stream()
+ .mapToInt(span -> getAttribute(span, SQL_CACHE_UPDATES))
+ .sum();
+
+ assertEquals(expCacheUpdates, cacheUpdates);
+ }
+
+ /**
+ * Checks whether parent span has a single child span with specified type.
+ *
+ * @param type Span type.
+ * @param parentSpan Parent span id.
+ * @return Id of the the child span.
+ */
+ protected SpanId checkChildSpan(SpanType type, SpanId parentSpan) {
+ return checkSpan(type, parentSpan,1, null).get(0);
+ }
+
+ /**
+ * Finds child spans with specified type and parent span.
+ *
+ * @param type Span type.
+ * @param parentSpanId Parent span id.
+ * @return Ids of the found spans.
+ */
+ protected List<SpanId> findChildSpans(SpanType type, SpanId parentSpanId) {
+ return handler().allSpans()
+ .filter(span -> parentSpanId != null ?
+ parentSpanId.equals(span.getParentSpanId()) && type.spanName().equals(span.getName()) :
+ type.spanName().equals(span.getName()))
+ .map(span -> span.getContext().getSpanId())
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Obtains integer value of the attribtute from span with specified id.
+ *
+ * @param spanId Id of the target span.
+ * @param tag Tag of the attribute.
+ * @return Value of the attribute.
+ */
+ protected int getAttribute(SpanId spanId, String tag) {
+ return parseInt(attributeValueToString(handler()
+ .spanById(spanId)
+ .getAttributes()
+ .getAttributeMap()
+ .get(tag)));
+ }
+
+ /**
+ * Executes query with specified parameters.
+ *
+ * @param sql SQL query to execute.
+ * @param schema SQL query schema.
+ * @param skipReducerOnUpdate Wether reduce phase should be skipped during update query execution.
+ * @param distributedJoins Whether distributed joins enabled.
+ * @param isQry {@code True} if query is SELECT, {@code False} - DML queries, {@code null} - all remainings.
+ */
+ protected void executeQuery(
+ String sql,
+ String schema,
+ boolean skipReducerOnUpdate,
+ boolean distributedJoins,
+ Boolean isQry
+ ) {
+ SqlFieldsQuery qry = new SqlFieldsQueryEx(sql, isQry)
+ .setSkipReducerOnUpdate(skipReducerOnUpdate)
+ .setDistributedJoins(distributedJoins)
+ .setPageSize(PAGE_SIZE)
+ .setSchema(schema);
+
+ reducer().context().query().querySqlFields(qry, false).getAll();
+ }
+
+ /**
+ * Executes specified query and checks the root span of query execution.
+ *
+ * @param sql SQL query to execute.
+ * @param schema SQL query schema.
+ * @param skipReducerOnUpdate Wether reduce phase should be skipped during update query execution.
+ * @param distributedJoins Whether distributed joins enabled.
+ * @param isQry {@code True} if query is SELECT, {@code False} - DML queries, {@code null} - all remainings.
+ */
+ protected SpanId executeAndCheckRootSpan(
+ String sql,
+ String schema,
+ boolean skipReducerOnUpdate,
+ boolean distributedJoins,
+ Boolean isQry
+ ) throws Exception {
+ executeQuery(sql, schema, skipReducerOnUpdate, distributedJoins, isQry);
+
+ handler().flush();
+
+ checkDroppedSpans();
+
+ return checkSpan(
+ SQL_QRY,
+ null,
+ 1,
+ ImmutableMap.<String, String>builder()
+ .put(NODE_ID, reducer().localNode().id().toString())
+ .put(tag(NODE, CONSISTENT_ID), reducer().localNode().consistentId().toString())
+ .put(tag(NODE, NAME), reducer().name())
+ .put(SQL_QRY_TEXT, sql)
+ .put(SQL_SCHEMA, schema)
+ .build()
+ ).get(0);
+ }
+
+ /**
+ * @return Name of the table which was created.
+ */
+ protected String createTableAndPopulate(Class<?> cls, CacheMode mode, int qryParallelism) {
+ IgniteCache<Integer, Object> cache = grid(0).createCache(
+ new CacheConfiguration<Integer, Object>(cls.getSimpleName() + "_" + mode)
+ .setIndexedTypes(Integer.class, cls)
+ .setCacheMode(mode)
+ .setQueryParallelism(qryParallelism)
+ .setSqlSchema(TEST_SCHEMA)
+ );
+
+ for (int i = 0; i < TEST_TABLE_POPULATION; i++)
+ cache.put(keyCntr.getAndIncrement(), cls == Organization.class ? new Organization(i, i) : new Person(i, i));
+
+ return TEST_SCHEMA + '.' + cls.getSimpleName();
+ }
+
+ /**
+ * Checks that no spans were dropped by OpencenCensus due to exporter buffer overflow.
+ */
+ private void checkDroppedSpans() {
+ Object worker = U.field(Tracing.getExportComponent().getSpanExporter(), "worker");
+
+ long droppedSpans = U.field(worker, "droppedSpans");
+
+ assertEquals("Some spans were dropped by OpencenCensus due to exporter buffer overflow.",
+ 0, droppedSpans);
+ }
+
+ /**
+ * @return Reducer node.
+ */
+ protected IgniteEx reducer() {
+ return ignite(GRID_CNT);
+ }
+
+ /**
+ * @return Number of nodes in cluster except reducer.
+ */
+ protected int mapNodesCount() {
+ return GRID_CNT;
+ }
+
+ /** */
+ public static class Person {
+ /** */
+ @QuerySqlField(index = true)
+ private int prsnId;
+
+ /** */
+ @QuerySqlField
+ private int prsnVal;
+
+ /** */
+ public Person(int prsnId, int prsnVal) {
+ this.prsnId = prsnId;
+ this.prsnVal = prsnVal;
+ }
+ }
+
+ /** */
+ public static class Organization {
+ /** */
+ @QuerySqlField(index = true)
+ private int orgId;
+
+ /** */
+ @QuerySqlField
+ private int orgVal;
+
+ /** */
+ public Organization(int orgId, int orgVal) {
+ this.orgId = orgId;
+ this.orgVal = orgVal;
+ }
+ }
+}