diff --git a/modules/clients/src/test/java/org/apache/ignite/common/RunningQueryInfoCheckInitiatorTest.java b/modules/clients/src/test/java/org/apache/ignite/common/RunningQueryInfoCheckInitiatorTest.java index c00e27cf547b5..86204c379c07e 100644 --- a/modules/clients/src/test/java/org/apache/ignite/common/RunningQueryInfoCheckInitiatorTest.java +++ b/modules/clients/src/test/java/org/apache/ignite/common/RunningQueryInfoCheckInitiatorTest.java @@ -336,7 +336,7 @@ private String initiatorId(IgniteEx node, String sqlMatch, int timeout) throws E fail("Timeout. Cannot find query with: " + sqlMatch); List> res = node.context().query().querySqlFields( - new SqlFieldsQuery("SELECT sql, initiator_id FROM SYS.SQL_QUERIES"), false).getAll(); + new SqlFieldsQuery("SELECT sql, initiator_id FROM SYS.SQL_QUERIES WHERE MAP_QUERY = FALSE"), false).getAll(); for (List row : res) { if (((String)row.get(0)).toUpperCase().contains(sqlMatch.toUpperCase())) @@ -356,7 +356,7 @@ private void checkRunningQueriesCount(IgniteEx node, int expectedQryCount, int t while (true) { List> res = node.context().query().querySqlFields( - new SqlFieldsQuery("SELECT * FROM SYS.SQL_QUERIES"), false).getAll(); + new SqlFieldsQuery("SELECT * FROM SYS.SQL_QUERIES WHERE MAP_QUERY = FALSE"), false).getAll(); res.stream().forEach(System.out::println); diff --git a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java index b2de892d55eff..176e1b49a47cd 100644 --- a/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java +++ b/modules/clients/src/test/java/org/apache/ignite/jdbc/thin/JdbcThinMetadataSelfTest.java @@ -785,8 +785,10 @@ public void testGetAllColumns() throws Exception { "SYS.SQL_QUERIES.LOCAL.null", "SYS.SQL_QUERIES.START_TIME.null", "SYS.SQL_QUERIES.DURATION.null", + "SYS.SQL_QUERIES.NODE_ID.null", "SYS.SQL_QUERIES.ORIGIN_NODE_ID.null", "SYS.SQL_QUERIES.INITIATOR_ID.null", + "SYS.SQL_QUERIES.MAP_QUERY.null", "SYS.SQL_QUERIES.SUBJECT_ID.null", "SYS.SCAN_QUERIES.START_TIME.null", "SYS.SCAN_QUERIES.TRANSFORMER.null", diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/GridRunningQueryInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/GridRunningQueryInfo.java index 69530a59bcbbf..defcbaf23d14f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/GridRunningQueryInfo.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/GridRunningQueryInfo.java @@ -33,9 +33,12 @@ public class GridRunningQueryInfo { /** */ private final long id; - /** Originating Node ID. */ + /** Node that owns query. */ private final UUID nodeId; + /** Query coordinator node ID. */ + private final UUID originNodeId; + /** */ private final String qry; @@ -67,6 +70,9 @@ public class GridRunningQueryInfo { /** Originator. */ private final String qryInitiatorId; + /** Map query flag. */ + private final boolean mapQry; + /** Enforce join order flag. */ private final boolean enforceJoinOrder; @@ -80,7 +86,8 @@ public class GridRunningQueryInfo { * Constructor. * * @param id Query ID. - * @param nodeId Originating node ID. + * @param nodeId Node that owns query. + * @param originNodeId Query coordinator node ID. * @param qry Query text. * @param qryType Query type. * @param schemaName Schema name. @@ -89,6 +96,7 @@ public class GridRunningQueryInfo { * @param cancel Query cancel. * @param loc Local query flag. * @param qryInitiatorId Query's initiator identifier. + * @param mapQry Map query flag. * @param enforceJoinOrder Enforce join order flag. * @param distributedJoins Distributed joins flag. * @param subjId Subject ID. @@ -96,6 +104,7 @@ public class GridRunningQueryInfo { public GridRunningQueryInfo( long id, UUID nodeId, + UUID originNodeId, String qry, GridCacheQueryType qryType, String schemaName, @@ -104,12 +113,14 @@ public GridRunningQueryInfo( GridQueryCancel cancel, boolean loc, String qryInitiatorId, + boolean mapQry, boolean enforceJoinOrder, boolean distributedJoins, UUID subjId ) { this.id = id; this.nodeId = nodeId; + this.originNodeId = originNodeId; this.qry = qry; this.qryType = qryType; this.schemaName = schemaName; @@ -119,6 +130,7 @@ public GridRunningQueryInfo( this.loc = loc; this.span = MTC.span(); this.qryInitiatorId = qryInitiatorId; + this.mapQry = mapQry; this.enforceJoinOrder = enforceJoinOrder; this.distributedJoins = distributedJoins; this.subjId = subjId; @@ -203,12 +215,19 @@ public boolean local() { } /** - * @return Originating node ID. + * @return Node that owns query. */ public UUID nodeId() { return nodeId; } + /** + * @return Query coordinator node ID. + */ + public UUID originNodeId() { + return originNodeId; + } + /** * @return Span of the running query. */ @@ -224,6 +243,13 @@ public String queryInitiatorId() { return qryInitiatorId; } + /** + * @return {@code true} if query executes map phase. + */ + public boolean mapQuery() { + return mapQry; + } + /** * @return Distributed joins. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/RunningQueryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/RunningQueryManager.java index b037037cc88b4..d2550c64894e0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/RunningQueryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/query/running/RunningQueryManager.java @@ -287,6 +287,68 @@ public void start(GridSpinBusyLock busyLock) { public long register(String qry, GridCacheQueryType qryType, String schemaName, boolean loc, @Nullable GridQueryCancel cancel, String qryInitiatorId, boolean enforceJoinOrder, boolean distributedJoins) { + return register( + qry, + qryType, + schemaName, + loc, + cancel, + qryInitiatorId, + enforceJoinOrder, + distributedJoins, + localNodeId, + false + ); + } + + /** + * Registers map-side running query and returns an id associated with the query on the current node. + * + * @param qry Query text. + * @param schemaName Schema name. + * @param cancel Query cancel. + * @param qryInitiatorId Query initiator ID. + * @param originNodeId Query origin node ID. + * @param enforceJoinOrder Enforce join order flag. + * @param distributedJoins Distributed joins flag. + * @return Id of registered query. + */ + public long registerMapQuery( + String qry, + String schemaName, + @Nullable GridQueryCancel cancel, + String qryInitiatorId, + UUID originNodeId, + boolean enforceJoinOrder, + boolean distributedJoins + ) { + return register( + qry, + SQL_FIELDS, + schemaName, + false, + cancel, + qryInitiatorId, + enforceJoinOrder, + distributedJoins, + originNodeId, + true + ); + } + + /** Registers running query and returns an id associated with the query. */ + private long register( + String qry, + GridCacheQueryType qryType, + String schemaName, + boolean loc, + @Nullable GridQueryCancel cancel, + String qryInitiatorId, + boolean enforceJoinOrder, + boolean distributedJoins, + UUID originNodeId, + boolean mapQry + ) { long qryId = qryIdGen.incrementAndGet(); if (qryInitiatorId == null) @@ -295,6 +357,7 @@ public long register(String qry, GridCacheQueryType qryType, String schemaName, final GridRunningQueryInfo run = new GridRunningQueryInfo( qryId, localNodeId, + originNodeId, qry, qryType, schemaName, @@ -303,6 +366,7 @@ public long register(String qry, GridCacheQueryType qryType, String schemaName, cancel, loc, qryInitiatorId, + mapQry, enforceJoinOrder, distributedJoins, securitySubjectId(ctx) @@ -314,7 +378,7 @@ public long register(String qry, GridCacheQueryType qryType, String schemaName, run.span().addTag(SQL_QRY_ID, run::globalQueryId); - if (!qryStartedListeners.isEmpty()) { + if (!mapQry && !qryStartedListeners.isEmpty()) { GridQueryStartedInfo info = new GridQueryStartedInfo( run.id(), localNodeId, @@ -375,26 +439,28 @@ public void unregister(long qryId, @Nullable Throwable failReason) { if (failed) qrySpan.addTag(ERROR, failReason::getMessage); - //We need to collect query history and metrics only for SQL queries. if (isSqlQuery(qry)) { qry.runningFuture().onDone(); - qryHistTracker.collectHistory(qry, failed); + // We need to collect query history and metrics only for external SQL queries. + if (!qry.mapQuery()) { + qryHistTracker.collectHistory(qry, failed); - if (!failed) - successQrsCnt.increment(); - else { - failedQrsCnt.increment(); + 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(); + // 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(); + } } } - if (ctx.performanceStatistics().enabled() && qry.startTimeNanos() > 0) { + if (!qry.mapQuery() && ctx.performanceStatistics().enabled() && qry.startTimeNanos() > 0) { String flags = null; // Create string for flags with not default values. @@ -426,7 +492,7 @@ public void unregister(long qryId, @Nullable Throwable failReason) { !failed); } - if (!qryFinishedListeners.isEmpty()) { + if (!qry.mapQuery() && !qryFinishedListeners.isEmpty()) { GridQueryFinishedInfo info = new GridQueryFinishedInfo( qry.id(), localNodeId, @@ -553,7 +619,7 @@ public Collection runningQueries(long duration) { long curTime = U.currentTimeMillis(); for (GridRunningQueryInfo runningQryInfo : runs.values()) { - if (curTime - runningQryInfo.startTime() > duration) + if (!runningQryInfo.mapQuery() && curTime - runningQryInfo.startTime() > duration) res.add(runningQryInfo); } diff --git a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SqlQueryView.java b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SqlQueryView.java index cd5f2d7d91b4b..5d1b80e82e7ac 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SqlQueryView.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/systemview/view/SqlQueryView.java @@ -40,12 +40,18 @@ public SqlQueryView(GridRunningQueryInfo qry) { this.qry = qry; } - /** @return Origin query node. */ + /** @return Node that owns this query. */ @Order(2) - public UUID originNodeId() { + public UUID nodeId() { return qry.nodeId(); } + /** @return Origin query node. */ + @Order(3) + public UUID originNodeId() { + return qry.originNodeId(); + } + /** @return Query ID. */ @Order public String queryId() { @@ -64,23 +70,29 @@ public String schemaName() { } /** @return Query start time. */ - @Order(3) + @Order(4) public Date startTime() { return new Date(qry.startTime()); } /** @return Query duration. */ - @Order(4) + @Order(5) public long duration() { return U.currentTimeMillis() - qry.startTime(); } /** @return Query initiator ID. */ - @Order(7) + @Order(8) public String initiatorId() { return qry.queryInitiatorId(); } + /** @return {@code True} if query executes map phase. */ + @Order(9) + public boolean mapQuery() { + return qry.mapQuery(); + } + /** @return {@code True} if query is local. */ public boolean local() { return qry.local(); diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/MapH2QueryInfo.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/MapH2QueryInfo.java index a159d2ac2c948..dbb0cb0d629af 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/MapH2QueryInfo.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/MapH2QueryInfo.java @@ -30,11 +30,15 @@ public class MapH2QueryInfo extends H2QueryInfo { /** Segment. */ private final int segment; + /** Running query id. */ + private final long runningQryId; + /** * @param stmt Query statement. * @param sql Query statement. * @param nodeId Originator node id. * @param qryId Query id. + * @param runningQryId Running query id. * @param initiatorId Query initiator id. * @param reqId Request ID. * @param segment Segment. @@ -44,19 +48,28 @@ public MapH2QueryInfo( String sql, UUID nodeId, long qryId, + long runningQryId, String initiatorId, long reqId, int segment ) { super(QueryType.MAP, stmt, sql, nodeId, qryId, initiatorId); + this.runningQryId = runningQryId; this.reqId = reqId; this.segment = segment; } + /** @return Running query id. */ + public long runningQueryId() { + return runningQryId; + } + /** {@inheritDoc} */ @Override protected void printInfo(StringBuilder msg) { - msg.append(", reqId=").append(reqId) + msg.append(", mapQuery=true") + .append(", originNodeId=").append(nodeId()) + .append(", reqId=").append(reqId) .append(", segment=").append(segment); } } 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 f2a4b459a7d46..ce047688b102d 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 @@ -71,6 +71,7 @@ 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.query.running.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.Span; @@ -454,6 +455,8 @@ private void onQueryRequest0( MapH2QueryInfo qryInfo = null; + long runningQryId = RunningQueryManager.UNDEFINED_QUERY_ID; + try { res.lock(); @@ -468,7 +471,28 @@ private void onQueryRequest0( H2Utils.bindParameters(stmt, params0); - qryInfo = new MapH2QueryInfo(stmt, qry.query(), node.id(), qryId, qryInitiatorId, reqId, segmentId); + GridQueryCancel qryCancel = qryResults.queryCancel(qryIdx); + + runningQryId = h2.runningQueryManager().registerMapQuery( + sql, + schemaName, + qryCancel, + qryInitiatorId, + node.id(), + enforceJoinOrder, + distributedJoins + ); + + qryInfo = new MapH2QueryInfo( + stmt, + qry.query(), + node.id(), + qryId, + runningQryId, + qryInitiatorId, + reqId, + segmentId + ); h2.heavyQueriesTracker().startTracking(qryInfo); @@ -482,8 +506,6 @@ private void onQueryRequest0( ); } - GridQueryCancel qryCancel = qryResults.queryCancel(qryIdx); - ResultSet rs = h2.executeWithResumableTimeTracking( () -> h2.executeSqlQueryWithTimer( stmt, @@ -567,6 +589,8 @@ private void onQueryRequest0( if (qryInfo != null) h2.heavyQueriesTracker().stopTracking(qryInfo, e); + h2.runningQueryManager().unregister(runningQryId, e); + throw e; } finally { diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResult.java b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResult.java index 0d844d289f5f5..9b86603c3fca4 100644 --- a/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResult.java +++ b/modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/twostep/MapQueryResult.java @@ -390,6 +390,8 @@ void close() { U.close(rs, log); h2.heavyQueriesTracker().stopTracking(qryInfo, null); + + h2.runningQueryManager().unregister(qryInfo.runningQueryId(), null); } } } diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/LongRunningQueryTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/LongRunningQueryTest.java index 37c6edfcb8938..4cd18dbf33f29 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/LongRunningQueryTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/LongRunningQueryTest.java @@ -505,6 +505,31 @@ public void testQueryInitiatorId() { LONG_QUERY_WARNING_TIMEOUT); } + /** Verifies map query information in long-query logs. */ + @Test + @MultiNodeTest + public void testLongMapQueryLogInfo() { + String initiatorId = UUID.randomUUID().toString(); + + UUID originNodeId = ignite.cluster().localNode().id(); + + LogListener lsnr = LogListener.matches(LONG_QUERY_FINISHED_MSG) + .andMatches("type=MAP") + .andMatches("mapQuery=true") + .andMatches("originNodeId=" + originNodeId) + .andMatches("initiatorId=" + initiatorId) + .build(); + + testLog().registerListener(lsnr); + + ignite.cache("test").query(new SqlFieldsQuery("SELECT val FROM test WHERE id = sleep_func(?, 0)") + .setQueryInitiatorId(initiatorId) + .setArgs(LONG_QUERY_WARNING_TIMEOUT)) + .getAll(); + + assertTrue(lsnr.check()); + } + /** */ private void checkInitiatorId(ListeningTestLogger log, String type, String sql, Object... args) { String initiatorId = UUID.randomUUID().toString();