IGNITE-27313 SQL: Add information about query initiator id to log

Author: alex-plekhanov

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/integration/SqlDiagnosticIntegrationTest.java

@@ -1028,11 +1028,14 @@ public void testSqlFieldsQueryWithInitiatorId() throws Exception {
             }, 3_000));
         }
 
-        LogListener logLsnr = LogListener.matches(LONG_QUERY_FINISHED_MSG).andMatches("initiatorId=testId2").build();
+        String initiatorId = "testId2";
+
+        LogListener logLsnr = LogListener.matches(LONG_QUERY_FINISHED_MSG)
+            .andMatches("initiatorId=" + initiatorId).build();
 
         log.registerListener(logLsnr);
 
-        cache.query(new SqlFieldsQuery("SELECT sleep(?)").setArgs(LONG_QRY_TIMEOUT).setQueryInitiatorId("testId2"))
+        cache.query(new SqlFieldsQuery("SELECT sleep(?)").setArgs(LONG_QRY_TIMEOUT).setQueryInitiatorId(initiatorId))
             .getAll();
 
         assertTrue(logLsnr.check(1000));

modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/H2DmlInfo.java

@@ -38,6 +38,9 @@ public class H2DmlInfo implements TrackableQuery {
     /** Schema name. */
     private final String schema;
 
+    /** Query initiator id. */
+    private final String initiatorId;
+
     /** Dml command. */
     private final String sql;
 
@@ -48,11 +51,12 @@ public class H2DmlInfo implements TrackableQuery {
      * @param schema Schema name.
      * @param sql Dml command.
      */
-    public H2DmlInfo(long beginTs, long qryId, UUID initNodeId, String schema, String sql) {
+    public H2DmlInfo(long beginTs, long qryId, UUID initNodeId, String schema, String initiatorId, String sql) {
         this.beginTs = beginTs;
         this.qryId = qryId;
         this.initNodeId = initNodeId;
         this.schema = schema;
+        this.initiatorId = initiatorId;
         this.sql = sql;
     }
 
@@ -76,6 +80,7 @@ public H2DmlInfo(long beginTs, long qryId, UUID initNodeId, String schema, Strin
         msgSb.append(", duration=").append(time()).append("ms")
             .append(", type=DML")
             .append(", schema=").append(schema)
+            .append(", initiatorId=").append(initiatorId)
             .append(", sql='").append(sql).append("']");
 
         return msgSb.toString();

modules/indexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java

@@ -1064,6 +1064,7 @@ private List<? extends FieldsQueryCursor<List<?>>> executeDml(
                 qryId,
                 ctx.localNodeId(),
                 qryDesc.schemaName(),
+                qryDesc.queryInitiatorId(),
                 qryDesc.sql()
             );
 

modules/indexing/src/test/java/org/apache/ignite/internal/processors/query/LongRunningQueryTest.java

@@ -525,7 +525,7 @@ public void testEmptyHeavyQueriesTrackerWithMultipleCancelledQueries() {
      */
     @Test
     @MultiNodeTest
-    public void testQueryInitiatorId() throws Exception {
+    public void testQueryInitiatorId() {
         ListeningTestLogger testLog = testLog();
 
         checkInitiatorId(testLog, "LOCAL", "SELECT sleep_func(?, 0)", LONG_QUERY_WARNING_TIMEOUT);
@@ -535,13 +535,17 @@ public void testQueryInitiatorId() throws Exception {
 
         checkInitiatorId(testLog, "REDUCE", "SELECT sleep_func(?, sum(val)) FROM test WHERE id + 1 = 1",
             LONG_QUERY_WARNING_TIMEOUT);
+
+        checkInitiatorId(testLog, "DML", "UPDATE test SET val = sleep_func(?, val) WHERE id = 0",
+            LONG_QUERY_WARNING_TIMEOUT);
     }
 
     /** */
     private void checkInitiatorId(ListeningTestLogger log, String type, String sql, Object... args) {
         String initiatorId = UUID.randomUUID().toString();
 
-        LogListener lsnr = LogListener.matches(LONG_QUERY_FINISHED_MSG).andMatches(type).andMatches(initiatorId).build();
+        LogListener lsnr = LogListener.matches(LONG_QUERY_FINISHED_MSG).andMatches("type=" + type)
+            .andMatches("initiatorId=" + initiatorId).build();
 
         log.registerListener(lsnr);