Support system table current_queries and queries_costs_histogram

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java

@@ -137,4 +137,10 @@ public DataNodeConfig setDataNodeMemoryProportion(String dataNodeMemoryProportio
     setProperty("datanode_memory_proportion", dataNodeMemoryProportion);
     return this;
   }
+
+  @Override
+  public DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow) {
+    setProperty("query_cost_stat_window", String.valueOf(queryCostStatWindow));
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java

@@ -93,4 +93,9 @@ public DataNodeConfig setDeleteWalFilesPeriodInMs(long deleteWalFilesPeriodInMs)
   public DataNodeConfig setDataNodeMemoryProportion(String dataNodeMemoryProportion) {
     return this;
   }
+
+  @Override
+  public DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow) {
+    return this;
+  }
 }

integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java

@@ -51,4 +51,6 @@ DataNodeConfig setLoadTsFileAnalyzeSchemaMemorySizeInBytes(
   DataNodeConfig setDeleteWalFilesPeriodInMs(long deleteWalFilesPeriodInMs);
 
   DataNodeConfig setDataNodeMemoryProportion(String dataNodeMemoryProportion);
+
+  DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow);
 }

integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/informationschema/IoTDBCurrentQueriesIT.java

@@ -0,0 +1,205 @@
+/*
+ * 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.iotdb.relational.it.query.recent.informationschema;
+
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.queryengine.execution.QueryState;
+import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.framework.IoTDBTestRunner;
+import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
+import org.apache.iotdb.itbase.env.BaseEnv;
+
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.END_TIME_TABLE_MODEL;
+import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.NUMS;
+import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.STATEMENT_TABLE_MODEL;
+import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.STATE_TABLE_MODEL;
+import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.USER_TABLE_MODEL;
+import static org.apache.iotdb.commons.schema.table.InformationSchema.getSchemaTables;
+import static org.apache.iotdb.db.it.utils.TestUtils.createUser;
+import static org.apache.iotdb.itbase.env.BaseEnv.TABLE_SQL_DIALECT;
+import static org.junit.Assert.fail;
+
+@RunWith(IoTDBTestRunner.class)
+@Category({TableLocalStandaloneIT.class})
+// This IT will run at least 60s, so we only run it in 1C1D
+public class IoTDBCurrentQueriesIT {
+  private static final int CURRENT_QUERIES_COLUMN_NUM =
+      getSchemaTables().get("current_queries").getColumnNum();
+  private static final int QUERIES_COSTS_HISTOGRAM_COLUMN_NUM =
+      getSchemaTables().get("queries_costs_histogram").getColumnNum();
+  private static final String ADMIN_NAME =
+      CommonDescriptor.getInstance().getConfig().getDefaultAdminName();
+  private static final String ADMIN_PWD =
+      CommonDescriptor.getInstance().getConfig().getAdminPassword();
+
+  @BeforeClass
+  public static void setUp() throws Exception {
+    EnvFactory.getEnv().getConfig().getDataNodeConfig().setQueryCostStatWindow(1);
+    EnvFactory.getEnv().initClusterEnvironment();
+    createUser("test", "test123123456");
+  }
+
+  @AfterClass
+  public static void tearDown() throws Exception {
+    EnvFactory.getEnv().cleanClusterEnvironment();
+  }
+
+  @Test
+  public void testCurrentQueries() {
+    try {
+      Connection connection =
+          EnvFactory.getEnv().getConnection(ADMIN_NAME, ADMIN_PWD, BaseEnv.TABLE_SQL_DIALECT);
+      Statement statement = connection.createStatement();
+      statement.execute("USE information_schema");
+
+      // 1. query current_queries table
+      String sql = "SELECT * FROM current_queries";
+      ResultSet resultSet = statement.executeQuery(sql);
+      ResultSetMetaData metaData = resultSet.getMetaData();
+      Assert.assertEquals(CURRENT_QUERIES_COLUMN_NUM, metaData.getColumnCount());
+      int rowNum = 0;
+      while (resultSet.next()) {
+        Assert.assertEquals(QueryState.RUNNING.name(), resultSet.getString(STATE_TABLE_MODEL));
+        Assert.assertEquals(null, resultSet.getString(END_TIME_TABLE_MODEL));
+        Assert.assertEquals(sql, resultSet.getString(STATEMENT_TABLE_MODEL));
+        Assert.assertEquals(ADMIN_NAME, resultSet.getString(USER_TABLE_MODEL));
+        rowNum++;
+      }
+      Assert.assertEquals(1, rowNum);
+      resultSet.close();
+
+      // 2. query queries_costs_histogram table
+      sql = "SELECT * FROM queries_costs_histogram";
+      resultSet = statement.executeQuery(sql);
+      metaData = resultSet.getMetaData();
+      Assert.assertEquals(QUERIES_COSTS_HISTOGRAM_COLUMN_NUM, metaData.getColumnCount());
+      rowNum = 0;
+      int queriesCount = 0;
+      while (resultSet.next()) {
+        int nums = resultSet.getInt(NUMS);
+        if (nums > 0) {
+          queriesCount++;
+        }
+        rowNum++;
+      }
+      Assert.assertEquals(1, queriesCount);
+      Assert.assertEquals(61, rowNum);
+
+      // 3. requery current_queries table
+      sql = "SELECT * FROM current_queries";
+      resultSet = statement.executeQuery(sql);
+      metaData = resultSet.getMetaData();
+      Assert.assertEquals(CURRENT_QUERIES_COLUMN_NUM, metaData.getColumnCount());
+      rowNum = 0;
+      int finishedQueries = 0;
+      while (resultSet.next()) {
+        if (QueryState.FINISHED.name().equals(resultSet.getString(STATE_TABLE_MODEL))) {
+          finishedQueries++;
+        }
+        rowNum++;
+      }
+      // three rows in the result, 2 FINISHED and 1 RUNNING
+      Assert.assertEquals(3, rowNum);
+      Assert.assertEquals(2, finishedQueries);
+      resultSet.close();
+
+      // 4. test the expired QueryInfo was evicted
+      Thread.sleep(61_001);
+      resultSet = statement.executeQuery(sql);
+      rowNum = 0;
+      while (resultSet.next()) {
+        rowNum++;
+      }
+      // one row in the result, current query
+      Assert.assertEquals(1, rowNum);
+      resultSet.close();
+
+      sql = "SELECT * FROM queries_costs_histogram";
+      resultSet = statement.executeQuery(sql);
+      queriesCount = 0;
+      while (resultSet.next()) {
+        int nums = resultSet.getInt(NUMS);
+        if (nums > 0) {
+          queriesCount++;
+        }
+      }
+      // the last current_queries table query was recorded, others are evicted
+      Assert.assertEquals(1, queriesCount);
+    } catch (Exception e) {
+      fail(e.getMessage());
+    }
+
+    // 5. test privilege
+    testPrivilege();
+  }
+
+  private void testPrivilege() {
+    // 1. test current_queries table
+    try (Connection connection =
+            EnvFactory.getEnv().getConnection("test", "test123123456", TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      String sql = "SELECT * FROM information_schema.current_queries";
+
+      // another user executes a query
+      try (Connection connection2 =
+          EnvFactory.getEnv().getConnection(ADMIN_NAME, ADMIN_PWD, BaseEnv.TABLE_SQL_DIALECT)) {
+        ResultSet resultSet = connection2.createStatement().executeQuery(sql);
+        resultSet.close();
+      } catch (Exception e) {
+        fail(e.getMessage());
+      }
+
+      // current user query current_queries table
+      ResultSet resultSet = statement.executeQuery(sql);
+      int rowNum = 0;
+      while (resultSet.next()) {
+        rowNum++;
+      }
+      // only current query in the result
+      Assert.assertEquals(1, rowNum);
+    } catch (SQLException e) {
+      fail(e.getMessage());
+    }
+
+    // 2. test queries_costs_histogram table
+    try (Connection connection =
+            EnvFactory.getEnv().getConnection("test", "test123123456", TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.executeQuery("SELECT * FROM information_schema.queries_costs_histogram");
+    } catch (SQLException e) {
+      Assert.assertEquals(
+          "803: Access Denied: No permissions for this operation, please add privilege SYSTEM",
+          e.getMessage());
+    }
+  }
+}

integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java

@@ -399,6 +399,7 @@ public void testInformationSchema() throws SQLException {
               "config_nodes,INF,",
               "configurations,INF,",
               "connections,INF,",
+              "current_queries,INF,",
               "data_nodes,INF,",
               "databases,INF,",
               "functions,INF,",
@@ -407,6 +408,7 @@ public void testInformationSchema() throws SQLException {
               "pipe_plugins,INF,",
               "pipes,INF,",
               "queries,INF,",
+              "queries_costs_histogram,INF,",
               "regions,INF,",
               "subscriptions,INF,",
               "tables,INF,",
@@ -634,12 +636,14 @@ public void testInformationSchema() throws SQLException {
                   "information_schema,config_nodes,INF,USING,null,SYSTEM VIEW,",
                   "information_schema,data_nodes,INF,USING,null,SYSTEM VIEW,",
                   "information_schema,connections,INF,USING,null,SYSTEM VIEW,",
+                  "information_schema,current_queries,INF,USING,null,SYSTEM VIEW,",
+                  "information_schema,queries_costs_histogram,INF,USING,null,SYSTEM VIEW,",
                   "test,test,INF,USING,test,BASE TABLE,",
                   "test,view_table,100,USING,null,VIEW FROM TREE,")));
       TestUtils.assertResultSetEqual(
           statement.executeQuery("count devices from tables where status = 'USING'"),
           "count(devices),",
-          Collections.singleton("19,"));
+          Collections.singleton("21,"));
       TestUtils.assertResultSetEqual(
           statement.executeQuery(
               "select * from columns where table_name = 'queries' or database = 'test'"),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

@@ -814,6 +814,9 @@ public class IoTDBConfig {
   /** time cost(ms) threshold for slow query. Unit: millisecond */
   private long slowQueryThreshold = 10000;
 
+  /** time window threshold for record of history queries. Unit: minute */
+  private int queryCostStatWindow = 0;
+
   private int patternMatchingThreshold = 1000000;
 
   /**
@@ -2627,6 +2630,14 @@ public void setSlowQueryThreshold(long slowQueryThreshold) {
     this.slowQueryThreshold = slowQueryThreshold;
   }
 
+  public int getQueryCostStatWindow() {
+    return queryCostStatWindow;
+  }
+
+  public void setQueryCostStatWindow(int queryCostStatWindow) {
+    this.queryCostStatWindow = queryCostStatWindow;
+  }
+
   public boolean isEnableIndex() {
     return enableIndex;
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java

@@ -813,6 +813,11 @@ public void loadProperties(TrimProperties properties) throws BadNodeUrlException
             properties.getProperty(
                 "slow_query_threshold", String.valueOf(conf.getSlowQueryThreshold()))));
 
+    conf.setQueryCostStatWindow(
+        Integer.parseInt(
+            properties.getProperty(
+                "query_cost_stat_window", String.valueOf(conf.getQueryCostStatWindow()))));
+
     conf.setDataRegionNum(
         Integer.parseInt(
             properties.getProperty("data_region_num", String.valueOf(conf.getDataRegionNum()))));

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java

@@ -212,7 +212,8 @@ public Response executeFastLastQueryStatement(
       t = e;
       return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build();
     } finally {
-      long costTime = System.nanoTime() - startTime;
+      long endTime = System.nanoTime();
+      long costTime = endTime - startTime;
 
       StatementType statementType =
           Optional.ofNullable(statement)
@@ -227,7 +228,18 @@ public Response executeFastLastQueryStatement(
       if (queryId != null) {
         COORDINATOR.cleanupQueryExecution(queryId);
       } else {
-        recordQueries(() -> costTime, new FastLastQueryContentSupplier(prefixPathList), t);
+        IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+
+        Supplier<String> contentOfQuerySupplier = new FastLastQueryContentSupplier(prefixPathList);
+        COORDINATOR.recordCurrentQueries(
+            null,
+            startTime / 1_000_000,
+            endTime / 1_000_000,
+            costTime,
+            contentOfQuerySupplier,
+            clientSession.getUsername(),
+            clientSession.getClientAddress());
+        recordQueries(() -> costTime, contentOfQuerySupplier, t);
       }
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java

@@ -1050,13 +1050,23 @@ public TSExecuteStatementResp executeFastLastDataQueryForOnePrefixPath(
 
       resp.setMoreData(false);
 
-      long costTime = System.nanoTime() - startTime;
+      long endTime = System.nanoTime();
+      long costTime = endTime - startTime;
 
       CommonUtils.addStatementExecutionLatency(
           OperationType.EXECUTE_QUERY_STATEMENT, StatementType.FAST_LAST_QUERY.name(), costTime);
       CommonUtils.addQueryLatency(StatementType.FAST_LAST_QUERY, costTime);
-      recordQueries(
-          () -> costTime, () -> String.format("thrift fastLastQuery %s", prefixPath), null);
+
+      String statement = String.format("thrift fastLastQuery %s", prefixPath);
+      COORDINATOR.recordCurrentQueries(
+          null,
+          startTime / 1_000_000,
+          endTime / 1_000_000,
+          costTime,
+          () -> statement,
+          clientSession.getUsername(),
+          clientSession.getClientAddress());
+      recordQueries(() -> costTime, () -> statement, null);
       return resp;
     } catch (final Exception e) {
       return RpcUtils.getTSExecuteStatementResp(

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/QueryId.java

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.queryengine.common;
 
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
 
 import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -37,6 +38,8 @@ public class QueryId {
 
   public static final QueryId MOCK_QUERY_ID = QueryId.valueOf("mock_query_id");
 
+  private static final int DATANODE_ID = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+
   private final String id;
 
   private int nextPlanNodeIndex;
@@ -67,6 +70,10 @@ public String getId() {
     return id;
   }
 
+  public static int getDataNodeId() {
+    return DATANODE_ID;
+  }
+
   @Override
   public String toString() {
     return id;