This is an automated email from the ASF dual-hosted git repository. jackietien pushed a commit to branch force_ci/object_type in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 2b12cb64454a5eb8b808afb571f9abaf5ff5046b Author: Weihao Li <[email protected]> AuthorDate: Wed Dec 10 19:36:32 2025 +0800 Support system table current_queries and queries_costs_histogram (#16890) (cherry picked from commit 94461b065095dd8499c85b8691d89da35a878299) --- .../it/env/cluster/config/MppDataNodeConfig.java | 6 + .../it/env/remote/config/RemoteDataNodeConfig.java | 5 + .../apache/iotdb/itbase/env/DataNodeConfig.java | 2 + .../informationschema/IoTDBCurrentQueriesIT.java | 205 ++++++++++++++ .../relational/it/schema/IoTDBDatabaseIT.java | 6 +- .../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 + .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 5 + .../protocol/rest/v2/impl/RestApiServiceImpl.java | 16 +- .../protocol/thrift/impl/ClientRPCServiceImpl.java | 16 +- .../iotdb/db/queryengine/common/QueryId.java | 7 + .../InformationSchemaContentSupplierFactory.java | 146 +++++++++- .../iotdb/db/queryengine/plan/Coordinator.java | 301 ++++++++++++++++++++- .../plan/execution/IQueryExecution.java | 2 + .../queryengine/plan/execution/QueryExecution.java | 5 + .../plan/execution/config/ConfigExecution.java | 4 + .../DataNodeLocationSupplierFactory.java | 2 + .../operator/MergeTreeSortOperatorTest.java | 5 + .../plan/relational/planner/PlanTester.java | 2 + .../informationschema/CurrentQueriesTest.java | 107 ++++++++ .../informationschema}/ShowQueriesTest.java | 8 +- .../conf/iotdb-system.properties.template | 6 + .../iotdb/commons/concurrent/ThreadName.java | 1 + .../schema/column/ColumnHeaderConstant.java | 10 +- .../commons/schema/table/InformationSchema.java | 41 ++- 24 files changed, 895 insertions(+), 24 deletions(-) diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java index 357c15c7856..5e418072a7d 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppDataNodeConfig.java @@ -137,4 +137,10 @@ public class MppDataNodeConfig extends MppBaseConfig implements DataNodeConfig { setProperty("datanode_memory_proportion", dataNodeMemoryProportion); return this; } + + @Override + public DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow) { + setProperty("query_cost_stat_window", String.valueOf(queryCostStatWindow)); + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java index 1af7cb8f613..bba4c964f95 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/it/env/remote/config/RemoteDataNodeConfig.java @@ -93,4 +93,9 @@ public class RemoteDataNodeConfig implements DataNodeConfig { public DataNodeConfig setDataNodeMemoryProportion(String dataNodeMemoryProportion) { return this; } + + @Override + public DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow) { + return this; + } } diff --git a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java index 0ae46ffc70f..d57015b1396 100644 --- a/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java +++ b/integration-test/src/main/java/org/apache/iotdb/itbase/env/DataNodeConfig.java @@ -51,4 +51,6 @@ public interface DataNodeConfig { DataNodeConfig setDeleteWalFilesPeriodInMs(long deleteWalFilesPeriodInMs); DataNodeConfig setDataNodeMemoryProportion(String dataNodeMemoryProportion); + + DataNodeConfig setQueryCostStatWindow(int queryCostStatWindow); } diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/informationschema/IoTDBCurrentQueriesIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/informationschema/IoTDBCurrentQueriesIT.java new file mode 100644 index 00000000000..66941ec784f --- /dev/null +++ b/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()); + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java index e7bab16ad1f..4736d9b0521 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/schema/IoTDBDatabaseIT.java @@ -399,6 +399,7 @@ public class IoTDBDatabaseIT { "config_nodes,INF,", "configurations,INF,", "connections,INF,", + "current_queries,INF,", "data_nodes,INF,", "databases,INF,", "functions,INF,", @@ -407,6 +408,7 @@ public class IoTDBDatabaseIT { "pipe_plugins,INF,", "pipes,INF,", "queries,INF,", + "queries_costs_histogram,INF,", "regions,INF,", "subscriptions,INF,", "tables,INF,", @@ -634,12 +636,14 @@ public class IoTDBDatabaseIT { "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'"), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java index 7591e9d3cab..3f19a2101a8 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java +++ b/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; /** @@ -2628,6 +2631,14 @@ public class IoTDBConfig { this.slowQueryThreshold = slowQueryThreshold; } + public int getQueryCostStatWindow() { + return queryCostStatWindow; + } + + public void setQueryCostStatWindow(int queryCostStatWindow) { + this.queryCostStatWindow = queryCostStatWindow; + } + public boolean isEnableIndex() { return enableIndex; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java index bd579b24c3f..e51f445c567 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java @@ -813,6 +813,11 @@ public class IoTDBDescriptor { 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())))); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java index 66be144edef..b657523987d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/RestApiServiceImpl.java @@ -212,7 +212,8 @@ public class RestApiServiceImpl extends RestApiService { 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 class RestApiServiceImpl extends RestApiService { 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); } } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index 07c2800799c..167a1fa914f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -1050,13 +1050,23 @@ public class ClientRPCServiceImpl implements IClientRPCServiceWithHandler { 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( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/QueryId.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/QueryId.java index a59ce8334bf..44e67aa7ba6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/QueryId.java +++ b/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 class QueryId { return id; } + public static int getDataNodeId() { + return DATANODE_ID; + } + @Override public String toString() { return id; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java index daceffce6b7..76d84c741de 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/InformationSchemaContentSupplierFactory.java @@ -68,6 +68,7 @@ import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.protocol.session.IClientSession; import org.apache.iotdb.db.protocol.session.SessionManager; import org.apache.iotdb.db.queryengine.common.ConnectionInfo; +import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.plan.Coordinator; import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution; import org.apache.iotdb.db.queryengine.plan.execution.config.metadata.relational.ShowCreateViewTask; @@ -166,6 +167,10 @@ public class InformationSchemaContentSupplierFactory { return new DataNodesSupplier(dataTypes, userEntity); case InformationSchema.CONNECTIONS: return new ConnectionsSupplier(dataTypes, userEntity); + case InformationSchema.CURRENT_QUERIES: + return new CurrentQueriesSupplier(dataTypes, userEntity); + case InformationSchema.QUERIES_COSTS_HISTOGRAM: + return new QueriesCostsHistogramSupplier(dataTypes, userEntity); default: throw new UnsupportedOperationException("Unknown table: " + tableName); } @@ -201,14 +206,11 @@ public class InformationSchemaContentSupplierFactory { final IQueryExecution queryExecution = queryExecutions.get(nextConsumedIndex); if (queryExecution.getSQLDialect().equals(IClientSession.SqlDialect.TABLE)) { - final String[] splits = queryExecution.getQueryId().split("_"); - final int dataNodeId = Integer.parseInt(splits[splits.length - 1]); - columnBuilders[0].writeBinary(BytesUtils.valueOf(queryExecution.getQueryId())); columnBuilders[1].writeLong( TimestampPrecisionUtils.convertToCurrPrecision( queryExecution.getStartExecutionTime(), TimeUnit.MILLISECONDS)); - columnBuilders[2].writeInt(dataNodeId); + columnBuilders[2].writeInt(QueryId.getDataNodeId()); columnBuilders[3].writeFloat( (float) (currTime - queryExecution.getStartExecutionTime()) / 1000); columnBuilders[4].writeBinary( @@ -1181,4 +1183,140 @@ public class InformationSchemaContentSupplierFactory { return sessionConnectionIterator.hasNext(); } } + + private static class CurrentQueriesSupplier extends TsBlockSupplier { + private int nextConsumedIndex; + private List<Coordinator.StatedQueriesInfo> queriesInfo; + + private CurrentQueriesSupplier(final List<TSDataType> dataTypes, final UserEntity userEntity) { + super(dataTypes); + queriesInfo = Coordinator.getInstance().getCurrentQueriesInfo(); + try { + accessControl.checkUserGlobalSysPrivilege(userEntity); + } catch (final AccessDeniedException e) { + queriesInfo = + queriesInfo.stream() + .filter(iQueryInfo -> userEntity.getUsername().equals(iQueryInfo.getUser())) + .collect(Collectors.toList()); + } + } + + @Override + protected void constructLine() { + final Coordinator.StatedQueriesInfo queryInfo = queriesInfo.get(nextConsumedIndex); + columnBuilders[0].writeBinary(BytesUtils.valueOf(queryInfo.getQueryId())); + columnBuilders[1].writeBinary(BytesUtils.valueOf(queryInfo.getQueryState())); + columnBuilders[2].writeLong( + TimestampPrecisionUtils.convertToCurrPrecision( + queryInfo.getStartTime(), TimeUnit.MILLISECONDS)); + if (queryInfo.getEndTime() == Coordinator.QueryInfo.DEFAULT_END_TIME) { + columnBuilders[3].appendNull(); + } else { + columnBuilders[3].writeLong( + TimestampPrecisionUtils.convertToCurrPrecision( + queryInfo.getEndTime(), TimeUnit.MILLISECONDS)); + } + columnBuilders[4].writeInt(QueryId.getDataNodeId()); + columnBuilders[5].writeFloat(queryInfo.getCostTime()); + columnBuilders[6].writeBinary(BytesUtils.valueOf(queryInfo.getStatement())); + columnBuilders[7].writeBinary(BytesUtils.valueOf(queryInfo.getUser())); + columnBuilders[8].writeBinary(BytesUtils.valueOf(queryInfo.getClientHost())); + resultBuilder.declarePosition(); + nextConsumedIndex++; + } + + @Override + public boolean hasNext() { + return nextConsumedIndex < queriesInfo.size(); + } + } + + private static class QueriesCostsHistogramSupplier extends TsBlockSupplier { + private int nextConsumedIndex; + private static final Binary[] BUCKETS = + new Binary[] { + BytesUtils.valueOf("[0,1)"), + BytesUtils.valueOf("[1,2)"), + BytesUtils.valueOf("[2,3)"), + BytesUtils.valueOf("[3,4)"), + BytesUtils.valueOf("[4,5)"), + BytesUtils.valueOf("[5,6)"), + BytesUtils.valueOf("[6,7)"), + BytesUtils.valueOf("[7,8)"), + BytesUtils.valueOf("[8,9)"), + BytesUtils.valueOf("[9,10)"), + BytesUtils.valueOf("[10,11)"), + BytesUtils.valueOf("[11,12)"), + BytesUtils.valueOf("[12,13)"), + BytesUtils.valueOf("[13,14)"), + BytesUtils.valueOf("[14,15)"), + BytesUtils.valueOf("[15,16)"), + BytesUtils.valueOf("[16,17)"), + BytesUtils.valueOf("[17,18)"), + BytesUtils.valueOf("[18,19)"), + BytesUtils.valueOf("[19,20)"), + BytesUtils.valueOf("[20,21)"), + BytesUtils.valueOf("[21,22)"), + BytesUtils.valueOf("[22,23)"), + BytesUtils.valueOf("[23,24)"), + BytesUtils.valueOf("[24,25)"), + BytesUtils.valueOf("[25,26)"), + BytesUtils.valueOf("[26,27)"), + BytesUtils.valueOf("[27,28)"), + BytesUtils.valueOf("[28,29)"), + BytesUtils.valueOf("[29,30)"), + BytesUtils.valueOf("[30,31)"), + BytesUtils.valueOf("[31,32)"), + BytesUtils.valueOf("[32,33)"), + BytesUtils.valueOf("[33,34)"), + BytesUtils.valueOf("[34,35)"), + BytesUtils.valueOf("[35,36)"), + BytesUtils.valueOf("[36,37)"), + BytesUtils.valueOf("[37,38)"), + BytesUtils.valueOf("[38,39)"), + BytesUtils.valueOf("[39,40)"), + BytesUtils.valueOf("[40,41)"), + BytesUtils.valueOf("[41,42)"), + BytesUtils.valueOf("[42,43)"), + BytesUtils.valueOf("[43,44)"), + BytesUtils.valueOf("[44,45)"), + BytesUtils.valueOf("[45,46)"), + BytesUtils.valueOf("[46,47)"), + BytesUtils.valueOf("[47,48)"), + BytesUtils.valueOf("[48,49)"), + BytesUtils.valueOf("[49,50)"), + BytesUtils.valueOf("[50,51)"), + BytesUtils.valueOf("[51,52)"), + BytesUtils.valueOf("[52,53)"), + BytesUtils.valueOf("[53,54)"), + BytesUtils.valueOf("[54,55)"), + BytesUtils.valueOf("[55,56)"), + BytesUtils.valueOf("[56,57)"), + BytesUtils.valueOf("[57,58)"), + BytesUtils.valueOf("[58,59)"), + BytesUtils.valueOf("[59,60)"), + BytesUtils.valueOf("60+") + }; + private final int[] currentQueriesCostHistogram; + + private QueriesCostsHistogramSupplier( + final List<TSDataType> dataTypes, final UserEntity userEntity) { + super(dataTypes); + accessControl.checkUserGlobalSysPrivilege(userEntity); + currentQueriesCostHistogram = Coordinator.getInstance().getCurrentQueriesCostHistogram(); + } + + @Override + protected void constructLine() { + columnBuilders[0].writeBinary(BUCKETS[nextConsumedIndex]); + columnBuilders[1].writeInt(currentQueriesCostHistogram[nextConsumedIndex]); + resultBuilder.declarePosition(); + nextConsumedIndex++; + } + + @Override + public boolean hasNext() { + return nextConsumedIndex < 61; + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java index 7708f6c18cd..78cdee720da 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/Coordinator.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.client.async.AsyncDataNodeInternalServiceClient; import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient; import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory; import org.apache.iotdb.commons.concurrent.ThreadName; +import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil; import org.apache.iotdb.commons.conf.CommonConfig; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.conf.IoTDBConstant; @@ -42,6 +43,7 @@ import org.apache.iotdb.db.queryengine.common.MPPQueryContext; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.common.SessionInfo; import org.apache.iotdb.db.queryengine.execution.QueryIdGenerator; +import org.apache.iotdb.db.queryengine.execution.QueryState; import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher; import org.apache.iotdb.db.queryengine.plan.analyze.lock.DataNodeSchemaLockManager; import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher; @@ -141,23 +143,37 @@ import org.apache.iotdb.db.queryengine.plan.statement.Statement; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.thrift.TBase; +import org.apache.tsfile.utils.Accountable; +import org.apache.tsfile.utils.RamUsageEstimator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.BlockingDeque; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiFunction; import java.util.function.LongSupplier; import java.util.function.Supplier; +import java.util.stream.Collectors; import static org.apache.iotdb.commons.utils.StatusUtils.needRetry; +import static org.apache.iotdb.db.queryengine.plan.Coordinator.QueryInfo.DEFAULT_END_TIME; import static org.apache.iotdb.db.utils.CommonUtils.getContentOfRequest; +import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance; +import static org.apache.tsfile.utils.RamUsageEstimator.sizeOfCharArray; /** * The coordinator for MPP. It manages all the queries which are executed in current Node. And it @@ -203,12 +219,32 @@ public class Coordinator { private final ConcurrentHashMap<Long, IQueryExecution> queryExecutionMap; + private final BlockingDeque<QueryInfo> currentQueriesInfo = new LinkedBlockingDeque<>(); + private final AtomicInteger[] currentQueriesCostHistogram = new AtomicInteger[61]; + private final ScheduledExecutorService retryFailTasksExecutor = + IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor( + ThreadName.EXPIRED_QUERIES_INFO_CLEAR.getName()); + private final StatementRewrite statementRewrite; private final List<PlanOptimizer> logicalPlanOptimizers; private final List<PlanOptimizer> distributionPlanOptimizers; private final DataNodeLocationSupplierFactory.DataNodeLocationSupplier dataNodeLocationSupplier; private final TypeManager typeManager; + { + for (int i = 0; i < 61; i++) { + currentQueriesCostHistogram[i] = new AtomicInteger(); + } + + ScheduledExecutorUtil.safelyScheduleWithFixedDelay( + retryFailTasksExecutor, + this::clearExpiredQueriesInfoTask, + 1_000L, + 1_000L, + TimeUnit.MILLISECONDS); + LOGGER.info("Expired-Queries-Info-Clear thread is successfully started."); + } + static { coordinatorMemoryBlock = IoTDBDescriptor.getInstance() @@ -625,12 +661,22 @@ public class Coordinator { try (SetThreadName threadName = new SetThreadName(queryExecution.getQueryId())) { LOGGER.debug("[CleanUpQuery]]"); queryExecution.stopAndCleanup(t); + boolean isUserQuery = queryExecution.isQuery() && queryExecution.isUserQuery(); + Supplier<String> contentOfQuerySupplier = + new ContentOfQuerySupplier(nativeApiRequest, queryExecution); + if (isUserQuery) { + recordCurrentQueries( + queryExecution.getQueryId(), + queryExecution.getStartExecutionTime(), + System.currentTimeMillis(), + queryExecution.getTotalExecutionTime(), + contentOfQuerySupplier, + queryExecution.getUser(), + queryExecution.getClientHostname()); + } queryExecutionMap.remove(queryId); - if (queryExecution.isQuery() && queryExecution.isUserQuery()) { - recordQueries( - queryExecution::getTotalExecutionTime, - new ContentOfQuerySupplier(nativeApiRequest, queryExecution), - t); + if (isUserQuery) { + recordQueries(queryExecution::getTotalExecutionTime, contentOfQuerySupplier, t); } } } @@ -722,4 +768,249 @@ public class Coordinator { public ExecutorService getDispatchExecutor() { return dispatchExecutor; } + + /** record query info in memory data structure */ + public void recordCurrentQueries( + String queryId, + long startTime, + long endTime, + long costTimeInNs, + Supplier<String> contentOfQuerySupplier, + String user, + String clientHost) { + if (CONFIG.getQueryCostStatWindow() <= 0) { + return; + } + + if (queryId == null) { + // fast Last query API executeFastLastDataQueryForOnePrefixPath will enter this + queryId = queryIdGenerator.createNextQueryId().getId(); + } + + // ns -> s + float costTimeInSeconds = costTimeInNs * 1e-9f; + + QueryInfo queryInfo = + new QueryInfo( + queryId, + startTime, + endTime, + costTimeInSeconds, + contentOfQuerySupplier.get(), + user, + clientHost); + + while (!coordinatorMemoryBlock.allocate(RamUsageEstimator.sizeOfObject(queryInfo))) { + // try to release memory from the head of queue + QueryInfo queryInfoToRelease = currentQueriesInfo.poll(); + if (queryInfoToRelease == null) { + // no element in the queue and the memory is still not enough, skip this record + return; + } else { + // release memory and unrecord in histogram + coordinatorMemoryBlock.release(RamUsageEstimator.sizeOfObject(queryInfoToRelease)); + unrecordInHistogram(queryInfoToRelease.costTime); + } + } + + currentQueriesInfo.addLast(queryInfo); + recordInHistogram(costTimeInSeconds); + } + + private void recordInHistogram(float costTimeInSeconds) { + int bucket = (int) costTimeInSeconds; + if (bucket < 60) { + currentQueriesCostHistogram[bucket].getAndIncrement(); + } else { + currentQueriesCostHistogram[60].getAndIncrement(); + } + } + + private void unrecordInHistogram(float costTimeInSeconds) { + int bucket = (int) costTimeInSeconds; + if (bucket < 60) { + currentQueriesCostHistogram[bucket].getAndDecrement(); + } else { + currentQueriesCostHistogram[60].getAndDecrement(); + } + } + + private void clearExpiredQueriesInfoTask() { + int queryCostStatWindow = CONFIG.getQueryCostStatWindow(); + if (queryCostStatWindow <= 0) { + return; + } + + // the QueryInfo smaller than expired time will be cleared + long expiredTime = System.currentTimeMillis() - queryCostStatWindow * 60 * 1_000L; + // peek head, the head QueryInfo is in the time window, return directly + QueryInfo queryInfo = currentQueriesInfo.peekFirst(); + if (queryInfo == null || queryInfo.endTime >= expiredTime) { + return; + } + + queryInfo = currentQueriesInfo.poll(); + while (queryInfo != null) { + if (queryInfo.endTime < expiredTime) { + // out of time window, clear queryInfo + coordinatorMemoryBlock.release(RamUsageEstimator.sizeOfObject(queryInfo)); + unrecordInHistogram(queryInfo.costTime); + queryInfo = currentQueriesInfo.poll(); + } else { + // the head of the queue is not expired, add back + currentQueriesInfo.addFirst(queryInfo); + // there is no more candidate to clear + return; + } + } + } + + public List<StatedQueriesInfo> getCurrentQueriesInfo() { + List<IQueryExecution> runningQueries = getAllQueryExecutions(); + Set<String> runningQueryIdSet = + runningQueries.stream().map(IQueryExecution::getQueryId).collect(Collectors.toSet()); + List<StatedQueriesInfo> result = new ArrayList<>(); + + // add History queries (satisfy the time window) info + Iterator<QueryInfo> historyQueriesIterator = currentQueriesInfo.iterator(); + Set<String> repetitionQueryIdSet = new HashSet<>(); + long currentTime = System.currentTimeMillis(); + long needRecordTime = currentTime - CONFIG.getQueryCostStatWindow() * 60 * 1_000L; + while (historyQueriesIterator.hasNext()) { + QueryInfo queryInfo = historyQueriesIterator.next(); + if (queryInfo.endTime < needRecordTime) { + // out of time window, ignore it + } else { + if (runningQueryIdSet.contains(queryInfo.queryId)) { + repetitionQueryIdSet.add(queryInfo.queryId); + } + result.add(new StatedQueriesInfo(QueryState.FINISHED, queryInfo)); + } + } + + // add Running queries info after remove the repetitions which has recorded in History queries + result.addAll( + runningQueries.stream() + .filter(queryExecution -> !repetitionQueryIdSet.contains(queryExecution.getQueryId())) + .map( + queryExecution -> + new StatedQueriesInfo( + QueryState.RUNNING, + queryExecution.getQueryId(), + queryExecution.getStartExecutionTime(), + DEFAULT_END_TIME, + (currentTime - queryExecution.getStartExecutionTime()) / 1000, + queryExecution.getExecuteSQL().orElse("UNKNOWN"), + queryExecution.getUser(), + queryExecution.getClientHostname())) + .collect(Collectors.toList())); + return result; + } + + public int[] getCurrentQueriesCostHistogram() { + return Arrays.stream(currentQueriesCostHistogram).mapToInt(AtomicInteger::get).toArray(); + } + + public static class QueryInfo implements Accountable { + public static final long DEFAULT_END_TIME = -1L; + private static final long INSTANCE_SIZE = shallowSizeOfInstance(QueryInfo.class); + + private final String queryId; + + // unit: millisecond + private final long startTime; + private final long endTime; + // unit: second + private final float costTime; + + private final String statement; + private final String user; + private final String clientHost; + + public QueryInfo( + String queryId, + long startTime, + long endTime, + float costTime, + String statement, + String user, + String clientHost) { + this.queryId = queryId; + this.startTime = startTime; + this.endTime = endTime; + this.costTime = costTime; + this.statement = statement; + this.user = user; + this.clientHost = clientHost; + } + + public String getClientHost() { + return clientHost; + } + + public String getUser() { + return user; + } + + public long getStartTime() { + return startTime; + } + + public long getEndTime() { + return endTime; + } + + public float getCostTime() { + return costTime; + } + + public String getQueryId() { + return queryId; + } + + public String getStatement() { + return statement; + } + + @Override + public long ramBytesUsed() { + return INSTANCE_SIZE + + sizeOfCharArray(statement.length()) + + sizeOfCharArray(user.length()) + + sizeOfCharArray(clientHost.length()); + } + } + + public static class StatedQueriesInfo extends QueryInfo { + private final QueryState queryState; + + private StatedQueriesInfo(QueryState queryState, QueryInfo queryInfo) { + super( + queryInfo.queryId, + queryInfo.startTime, + queryInfo.endTime, + queryInfo.costTime, + queryInfo.statement, + queryInfo.user, + queryInfo.clientHost); + this.queryState = queryState; + } + + private StatedQueriesInfo( + QueryState queryState, + String queryId, + long startTime, + long endTime, + long costTime, + String statement, + String user, + String clientHost) { + super(queryId, startTime, endTime, costTime, statement, user, clientHost); + this.queryState = queryState; + } + + public String getQueryState() { + return queryState.name(); + } + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/IQueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/IQueryExecution.java index 98257c24293..e98f016767f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/IQueryExecution.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/IQueryExecution.java @@ -77,4 +77,6 @@ public interface IQueryExecution { IClientSession.SqlDialect getSQLDialect(); String getUser(); + + String getClientHostname(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java index 2c1657e839e..4734db5850a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java @@ -700,6 +700,11 @@ public class QueryExecution implements IQueryExecution { return context.getSession().getUserName(); } + @Override + public String getClientHostname() { + return context.getCliHostname(); + } + public MPPQueryContext getContext() { return context; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigExecution.java index 1880924f297..823a620820f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigExecution.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/ConfigExecution.java @@ -353,4 +353,8 @@ public class ConfigExecution implements IQueryExecution { public String getUser() { return context.getSession().getUserName(); } + + public String getClientHostname() { + return context.getCliHostname(); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java index f8cf497546e..d7d755ddc1d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/DataNodeLocationSupplierFactory.java @@ -86,6 +86,8 @@ public class DataNodeLocationSupplierFactory { switch (tableName) { case InformationSchema.QUERIES: case InformationSchema.CONNECTIONS: + case InformationSchema.CURRENT_QUERIES: + case InformationSchema.QUERIES_COSTS_HISTOGRAM: return getReadableDataNodeLocations(); case InformationSchema.DATABASES: case InformationSchema.TABLES: diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeTreeSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeTreeSortOperatorTest.java index 4ca38a5c8d9..8fc7d01437c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeTreeSortOperatorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeTreeSortOperatorTest.java @@ -1910,5 +1910,10 @@ public class MergeTreeSortOperatorTest { public boolean isUserQuery() { return false; } + + @Override + public String getClientHostname() { + return SessionConfig.DEFAULT_HOST; + } } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/PlanTester.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/PlanTester.java index a4cef177d81..85075570201 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/PlanTester.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/PlanTester.java @@ -83,6 +83,8 @@ public class PlanTester { public List<TDataNodeLocation> getDataNodeLocations(String table) { switch (table) { case "queries": + case "current_queries": + case "queries_costs_histogram": return ImmutableList.of( genDataNodeLocation(1, "192.0.1.1"), genDataNodeLocation(2, "192.0.1.2")); default: diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/CurrentQueriesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/CurrentQueriesTest.java new file mode 100644 index 00000000000..1e3163321ec --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/CurrentQueriesTest.java @@ -0,0 +1,107 @@ +/* + * 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.db.queryengine.plan.relational.planner.informationschema; + +import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan; +import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester; + +import com.google.common.collect.ImmutableList; +import org.junit.Test; + +import java.util.Optional; + +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.BIN; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.CLIENT_IP; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.COST_TIME; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.DATA_NODE_ID_TABLE_MODEL; +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.QUERY_ID_TABLE_MODEL; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.START_TIME_TABLE_MODEL; +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.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.collect; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.exchange; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.infoSchemaTableScan; +import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.output; + +public class CurrentQueriesTest { + private final PlanTester planTester = new PlanTester(); + + @Test + public void testCurrentQueries() { + LogicalQueryPlan logicalQueryPlan = + planTester.createPlan("select * from information_schema.current_queries"); + assertPlan( + logicalQueryPlan, + output( + infoSchemaTableScan( + "information_schema.current_queries", + Optional.empty(), + ImmutableList.of( + QUERY_ID_TABLE_MODEL, + STATE_TABLE_MODEL, + START_TIME_TABLE_MODEL, + END_TIME_TABLE_MODEL, + DATA_NODE_ID_TABLE_MODEL, + COST_TIME, + STATEMENT_TABLE_MODEL, + USER_TABLE_MODEL, + CLIENT_IP)))); + + // - Exchange + // Output - Collect - Exchange + assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange()))); + // TableScan + assertPlan( + planTester.getFragmentPlan(1), + infoSchemaTableScan("information_schema.current_queries", Optional.of(1))); + // TableScan + assertPlan( + planTester.getFragmentPlan(2), + infoSchemaTableScan("information_schema.current_queries", Optional.of(2))); + } + + @Test + public void testQueriesCostsHistogram() { + LogicalQueryPlan logicalQueryPlan = + planTester.createPlan("select * from information_schema.queries_costs_histogram"); + assertPlan( + logicalQueryPlan, + output( + infoSchemaTableScan( + "information_schema.queries_costs_histogram", + Optional.empty(), + ImmutableList.of(BIN, NUMS)))); + + // - Exchange + // Output - Collect - Exchange + assertPlan(planTester.getFragmentPlan(0), output(collect(exchange(), exchange()))); + // TableScan + assertPlan( + planTester.getFragmentPlan(1), + infoSchemaTableScan("information_schema.queries_costs_histogram", Optional.of(1))); + // TableScan + assertPlan( + planTester.getFragmentPlan(2), + infoSchemaTableScan("information_schema.queries_costs_histogram", Optional.of(2))); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ShowQueriesTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/ShowQueriesTest.java similarity index 94% rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ShowQueriesTest.java rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/ShowQueriesTest.java index 63b7b783d74..7161c68f4e9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/ShowQueriesTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/informationschema/ShowQueriesTest.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.iotdb.db.queryengine.plan.relational.analyzer; +package org.apache.iotdb.db.queryengine.plan.relational.planner.informationschema; import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan; import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester; @@ -32,7 +32,9 @@ import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.ELAPSE import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.QUERY_ID_TABLE_MODEL; import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.START_TIME_TABLE_MODEL; import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.STATEMENT; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.STATEMENT_TABLE_MODEL; import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.USER; +import static org.apache.iotdb.commons.schema.column.ColumnHeaderConstant.USER_TABLE_MODEL; import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan; import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.collect; import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.exchange; @@ -61,8 +63,8 @@ public class ShowQueriesTest { START_TIME_TABLE_MODEL, DATA_NODE_ID_TABLE_MODEL, ELAPSED_TIME_TABLE_MODEL, - STATEMENT.toLowerCase(Locale.ENGLISH), - USER.toLowerCase(Locale.ENGLISH))))); + STATEMENT_TABLE_MODEL, + USER_TABLE_MODEL)))); // - Exchange // Output - Collect - Exchange diff --git a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template index 5a32d6a1231..0a9a47ee890 100644 --- a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template +++ b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template @@ -1080,6 +1080,12 @@ max_tsblock_line_number=1000 # Datatype: long slow_query_threshold=10000 +# Time window threshold(min) for record of history queries. +# effectiveMode: hot_reload +# Datatype: int +# Privilege: SYSTEM +query_cost_stat_window=0 + # The max executing time of query. unit: ms # effectiveMode: restart # Datatype: int diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java index 390e9f80e9b..6f9f95ca8fe 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/concurrent/ThreadName.java @@ -35,6 +35,7 @@ public enum ThreadName { FRAGMENT_INSTANCE_NOTIFICATION("Fragment-Instance-Notification"), FRAGMENT_INSTANCE_DISPATCH("Fragment-Instance-Dispatch"), DRIVER_TASK_SCHEDULER_NOTIFICATION("Driver-Task-Scheduler-Notification"), + EXPIRED_QUERIES_INFO_CLEAR("Expired-Queries-Info-Clear"), // -------------------------- MPP -------------------------- MPP_COORDINATOR_SCHEDULED_EXECUTOR("MPP-Coordinator-Scheduled-Executor"), MPP_DATA_EXCHANGE_TASK_EXECUTOR("MPP-Data-Exchange-Task-Executors"), diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java index 67850d991cb..0459d4d2c86 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/column/ColumnHeaderConstant.java @@ -214,11 +214,19 @@ public class ColumnHeaderConstant { public static final String CLIENT_IP = "client_ip"; public static final String QUERY_ID_TABLE_MODEL = "query_id"; - public static final String QUERY_ID_START_TIME_TABLE_MODEL = "start_time"; public static final String DATA_NODE_ID_TABLE_MODEL = "datanode_id"; public static final String START_TIME_TABLE_MODEL = "start_time"; public static final String ELAPSED_TIME_TABLE_MODEL = "elapsed_time"; + // column names for current_queries and queries_costs_histogram + public static final String STATE_TABLE_MODEL = "state"; + public static final String END_TIME_TABLE_MODEL = "end_time"; + public static final String COST_TIME = "cost_time"; + public static final String STATEMENT_TABLE_MODEL = "statement"; + public static final String USER_TABLE_MODEL = "user"; + public static final String BIN = "bin"; + public static final String NUMS = "nums"; + public static final String TABLE_NAME_TABLE_MODEL = "table_name"; public static final String TABLE_TYPE_TABLE_MODEL = "table_type"; public static final String COLUMN_NAME_TABLE_MODEL = "column_name"; diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java index 243bc41c40c..b8e03423d61 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/schema/table/InformationSchema.java @@ -50,6 +50,8 @@ public class InformationSchema { public static final String CONFIG_NODES = "config_nodes"; public static final String DATA_NODES = "data_nodes"; public static final String CONNECTIONS = "connections"; + public static final String CURRENT_QUERIES = "current_queries"; + public static final String QUERIES_COSTS_HISTOGRAM = "queries_costs_histogram"; static { final TsTable queriesTable = new TsTable(QUERIES); @@ -57,17 +59,15 @@ public class InformationSchema { new TagColumnSchema(ColumnHeaderConstant.QUERY_ID_TABLE_MODEL, TSDataType.STRING)); queriesTable.addColumnSchema( new AttributeColumnSchema( - ColumnHeaderConstant.QUERY_ID_START_TIME_TABLE_MODEL, TSDataType.TIMESTAMP)); + ColumnHeaderConstant.START_TIME_TABLE_MODEL, TSDataType.TIMESTAMP)); queriesTable.addColumnSchema( new AttributeColumnSchema(ColumnHeaderConstant.DATA_NODE_ID_TABLE_MODEL, TSDataType.INT32)); queriesTable.addColumnSchema( new AttributeColumnSchema(ColumnHeaderConstant.ELAPSED_TIME_TABLE_MODEL, TSDataType.FLOAT)); queriesTable.addColumnSchema( - new AttributeColumnSchema( - ColumnHeaderConstant.STATEMENT.toLowerCase(Locale.ENGLISH), TSDataType.STRING)); + new AttributeColumnSchema(ColumnHeaderConstant.STATEMENT_TABLE_MODEL, TSDataType.STRING)); queriesTable.addColumnSchema( - new AttributeColumnSchema( - ColumnHeaderConstant.USER.toLowerCase(Locale.ENGLISH), TSDataType.STRING)); + new AttributeColumnSchema(ColumnHeaderConstant.USER_TABLE_MODEL, TSDataType.STRING)); queriesTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME); schemaTables.put(QUERIES, queriesTable); @@ -361,6 +361,37 @@ public class InformationSchema { new AttributeColumnSchema(ColumnHeaderConstant.CLIENT_IP, TSDataType.STRING)); connectionsTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME); schemaTables.put(CONNECTIONS, connectionsTable); + + final TsTable currentQueriesTable = new TsTable(CURRENT_QUERIES); + currentQueriesTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.QUERY_ID_TABLE_MODEL, TSDataType.STRING)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.STATE_TABLE_MODEL, TSDataType.STRING)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema( + ColumnHeaderConstant.START_TIME_TABLE_MODEL, TSDataType.TIMESTAMP)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.END_TIME_TABLE_MODEL, TSDataType.TIMESTAMP)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.DATA_NODE_ID_TABLE_MODEL, TSDataType.INT32)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.COST_TIME, TSDataType.FLOAT)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.STATEMENT_TABLE_MODEL, TSDataType.STRING)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.USER_TABLE_MODEL, TSDataType.STRING)); + currentQueriesTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.CLIENT_IP, TSDataType.STRING)); + currentQueriesTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME); + schemaTables.put(CURRENT_QUERIES, currentQueriesTable); + + final TsTable queriesCostsHistogramTable = new TsTable(QUERIES_COSTS_HISTOGRAM); + queriesCostsHistogramTable.addColumnSchema( + new TagColumnSchema(ColumnHeaderConstant.BIN, TSDataType.STRING)); + queriesCostsHistogramTable.addColumnSchema( + new AttributeColumnSchema(ColumnHeaderConstant.NUMS, TSDataType.INT32)); + queriesCostsHistogramTable.removeColumnSchema(TsTable.TIME_COLUMN_NAME); + schemaTables.put(QUERIES_COSTS_HISTOGRAM, queriesCostsHistogramTable); } public static Map<String, TsTable> getSchemaTables() {
