IGNITE-28621 SQL: Add H2 map query details to system view and logs

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<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<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);
 

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",

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,13 +96,15 @@ 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.
      */
     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.
      */

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<GridRunningQueryInfo> 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);
         }
 

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();

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);
     }
 }