From 65f2e5f16ea18e1de7194672643ed10a719f43f8 Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Wed, 8 Jul 2026 12:15:44 +0800 Subject: [PATCH 1/2] Enforce REST query row limit --- .../main/openapi3/iotdb_rest_table_v1.yaml | 1 + .../src/main/openapi3/iotdb_rest_v1.yaml | 1 + .../src/main/openapi3/iotdb_rest_v2.yaml | 1 + .../apache/iotdb/rest/i18n/RestMessages.java | 3 + .../apache/iotdb/rest/i18n/RestMessages.java | 3 + .../protocol/handler/QueryRowLimitUtils.java | 66 ++++++ .../table/v1/handler/QueryDataSetHandler.java | 44 +--- .../table/v1/impl/RestApiServiceImpl.java | 9 +- .../v1/handler/QueryDataSetHandler.java | 114 +++++---- .../v1/impl/GrafanaApiServiceImpl.java | 19 +- .../protocol/v1/impl/RestApiServiceImpl.java | 8 +- .../v2/handler/QueryDataSetHandler.java | 114 +++++---- .../v2/impl/GrafanaApiServiceImpl.java | 19 +- .../protocol/v2/impl/RestApiServiceImpl.java | 13 +- .../handler/QueryRowLimitUtilsTest.java | 52 ++++ .../v1/handler/QueryDataSetHandlerTest.java | 222 ++++++++++++++++++ .../test/resources/iotdb-common.properties | 3 +- .../test/resources/iotdb-common.properties | 3 +- .../test/resources/iotdb-system.properties | 3 +- .../conf/iotdb-system.properties.template | 3 +- 20 files changed, 524 insertions(+), 177 deletions(-) create mode 100644 external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java create mode 100644 external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java create mode 100644 external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandlerTest.java diff --git a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_table_v1.yaml b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_table_v1.yaml index 36cec631b54c2..27e347881c53c 100644 --- a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_table_v1.yaml +++ b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_table_v1.yaml @@ -98,6 +98,7 @@ components: row_limit: type: integer format: int32 + description: Maximum rows to return. The effective limit is capped by rest_query_default_row_size_limit. InsertTabletRequest: title: InsertTabletRequest diff --git a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v1.yaml b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v1.yaml index 18080c693ab50..7149ddd1cd395 100644 --- a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v1.yaml +++ b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v1.yaml @@ -166,6 +166,7 @@ components: rowLimit: type: integer format: int32 + description: Maximum rows to return. The effective limit is capped by rest_query_default_row_size_limit. InsertTabletRequest: title: InsertTabletRequest diff --git a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v2.yaml b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v2.yaml index 0cae51bef239a..0ebb430b1b9eb 100644 --- a/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v2.yaml +++ b/external-service-impl/rest-openapi/src/main/openapi3/iotdb_rest_v2.yaml @@ -201,6 +201,7 @@ components: row_limit: type: integer format: int32 + description: Maximum rows to return. The effective limit is capped by rest_query_default_row_size_limit. PrefixPathList: title: PrefixPathList diff --git a/external-service-impl/rest/src/main/i18n/en/org/apache/iotdb/rest/i18n/RestMessages.java b/external-service-impl/rest/src/main/i18n/en/org/apache/iotdb/rest/i18n/RestMessages.java index 1b4bf36432cac..882bb16c55808 100644 --- a/external-service-impl/rest/src/main/i18n/en/org/apache/iotdb/rest/i18n/RestMessages.java +++ b/external-service-impl/rest/src/main/i18n/en/org/apache/iotdb/rest/i18n/RestMessages.java @@ -34,6 +34,9 @@ public final class RestMessages { public static final String SQL_SHOULD_NOT_BE_NULL = "sql should not be null"; public static final String ROW_LIMIT_SHOULD_BE_POSITIVE = "row_limit should be positive"; public static final String ROW_LIMIT_CAMEL_SHOULD_BE_POSITIVE = "rowLimit should be positive"; + public static final String + MESSAGE_DATASET_ROW_SIZE_EXCEEDED_THE_GIVEN_MAX_ROW_SIZE_ARG_7A3F6452 = + "Dataset row size exceeded the given max row size (%d)"; public static final String PREFIX_PATHS_NOT_NULL = "prefix_paths should not be null"; public static final String TIMESTAMPS_NOT_NULL = "timestamps should not be null"; public static final String IS_ALIGNED_NOT_NULL = "is_aligned should not be null"; diff --git a/external-service-impl/rest/src/main/i18n/zh/org/apache/iotdb/rest/i18n/RestMessages.java b/external-service-impl/rest/src/main/i18n/zh/org/apache/iotdb/rest/i18n/RestMessages.java index 23f3475cdc4cc..cb443b6ec974f 100644 --- a/external-service-impl/rest/src/main/i18n/zh/org/apache/iotdb/rest/i18n/RestMessages.java +++ b/external-service-impl/rest/src/main/i18n/zh/org/apache/iotdb/rest/i18n/RestMessages.java @@ -34,6 +34,9 @@ public final class RestMessages { public static final String SQL_SHOULD_NOT_BE_NULL = "sql 不能为空"; public static final String ROW_LIMIT_SHOULD_BE_POSITIVE = "row_limit 应为正数"; public static final String ROW_LIMIT_CAMEL_SHOULD_BE_POSITIVE = "rowLimit 应为正数"; + public static final String + MESSAGE_DATASET_ROW_SIZE_EXCEEDED_THE_GIVEN_MAX_ROW_SIZE_ARG_7A3F6452 = + "数据集行数超过给定的最大行数（%d）"; public static final String PREFIX_PATHS_NOT_NULL = "prefix_paths 不能为空"; public static final String TIMESTAMPS_NOT_NULL = "timestamps 不能为空"; public static final String IS_ALIGNED_NOT_NULL = "is_aligned 不能为空"; diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java new file mode 100644 index 0000000000000..4d8bd08b4f004 --- /dev/null +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java @@ -0,0 +1,66 @@ +/* + * 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.rest.protocol.handler; + +import org.apache.iotdb.rest.i18n.RestMessages; +import org.apache.iotdb.rest.protocol.model.ExecutionStatus; +import org.apache.iotdb.rpc.TSStatusCode; + +import jakarta.ws.rs.core.Response; + +public final class QueryRowLimitUtils { + + private static final int MIN_ROW_SIZE_LIMIT = 1; + + private QueryRowLimitUtils() {} + + public static int resolveActualRowSizeLimit( + Integer requestedRowSizeLimit, int configuredRowSizeLimit) { + int hardLimit = normalizeRowSizeLimit(configuredRowSizeLimit); + if (requestedRowSizeLimit == null) { + return hardLimit; + } + return normalizeRowSizeLimit(Math.min(requestedRowSizeLimit, hardLimit)); + } + + public static int normalizeRowSizeLimit(int rowSizeLimit) { + return Math.max(MIN_ROW_SIZE_LIMIT, rowSizeLimit); + } + + public static boolean exceedsLimit( + int fetchedRowCount, int incomingRowCount, int actualRowSizeLimit) { + return incomingRowCount > 0 + && (long) fetchedRowCount + incomingRowCount > normalizeRowSizeLimit(actualRowSizeLimit); + } + + public static Response buildRowSizeLimitExceededResponse(int actualRowSizeLimit) { + int rowSizeLimit = normalizeRowSizeLimit(actualRowSizeLimit); + return Response.ok() + .entity( + new ExecutionStatus() + .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) + .message( + String.format( + RestMessages + .MESSAGE_DATASET_ROW_SIZE_EXCEEDED_THE_GIVEN_MAX_ROW_SIZE_ARG_7A3F6452, + rowSizeLimit))) + .build(); + } +} diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/handler/QueryDataSetHandler.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/handler/QueryDataSetHandler.java index 4bc94553414a1..9067756d5c684 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/handler/QueryDataSetHandler.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/handler/QueryDataSetHandler.java @@ -22,9 +22,8 @@ import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement; import org.apache.iotdb.db.queryengine.common.header.DatasetHeader; import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution; -import org.apache.iotdb.rest.protocol.model.ExecutionStatus; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.table.v1.model.QueryDataSet; -import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.block.column.Column; import org.apache.tsfile.common.conf.TSFileConfig; @@ -45,7 +44,7 @@ public class QueryDataSetHandler { private QueryDataSetHandler() {} /** - * @param actualRowSizeLimit max number of rows to return. no limit when actualRowSizeLimit <= 0. + * @param actualRowSizeLimit max number of rows to return. */ public static Response fillQueryDataSet( IQueryExecution queryExecution, Statement statement, int actualRowSizeLimit) @@ -61,7 +60,6 @@ public static Response fillQueryDataSet( IQueryExecution queryExecution, final int actualRowSizeLimit) throws IoTDBException { QueryDataSet targetDataSet = new QueryDataSet(); int fetched = 0; - int columnNum = queryExecution.getOutputValueColumnCount(); DatasetHeader header = queryExecution.getDatasetHeader(); List resultColumns = header.getRespColumns(); @@ -73,17 +71,6 @@ public static Response fillQueryDataSet( targetDataSet.addValuesItem(new ArrayList<>()); } while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { if (fetched == 0) { @@ -94,22 +81,22 @@ public static Response fillQueryDataSet( } TsBlock tsBlock = optionalTsBlock.get(); int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } for (int k = 0; k < resultColumns.size(); k++) { Column column = tsBlock.getColumn(headerMap.get(resultColumns.get(k))); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(targetDataSet.getDataTypes(), k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } targetDataSet.setValues(convertColumnToRow(targetDataSet.getValues())); return Response.ok().entity(targetDataSet).build(); @@ -174,17 +161,6 @@ private static Response fillOtherQueryDataSet( int fetched = 0; int columnNum = queryExecution.getOutputValueColumnCount(); while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent()) { if (fetched == 0) { @@ -199,21 +175,21 @@ private static Response fillOtherQueryDataSet( targetDataSet.setValues(new ArrayList<>()); return Response.ok().entity(targetDataSet).build(); } + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } for (int k = 0; k < columnNum; k++) { Column column = tsBlock.getColumn(targetDataSetIndexToSourceDataSetIndex[k]); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(targetDataSet.getDataTypes(), k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } targetDataSet.setValues(convertColumnToRow(targetDataSet.getValues())); return Response.ok().entity(targetDataSet).build(); diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/impl/RestApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/impl/RestApiServiceImpl.java index 77197c3c088d8..a04ae7a16b4c3 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/impl/RestApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/table/v1/impl/RestApiServiceImpl.java @@ -38,6 +38,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement; import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.SetThreadName; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.table.v1.NotFoundException; import org.apache.iotdb.rest.protocol.table.v1.RestApiService; import org.apache.iotdb.rest.protocol.table.v1.handler.ExceptionHandler; @@ -63,11 +64,12 @@ public class RestApiServiceImpl extends RestApiService { private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); - private final Integer defaultQueryRowLimit; + private final int defaultQueryRowLimit; public RestApiServiceImpl() { defaultQueryRowLimit = - IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit(); + QueryRowLimitUtils.normalizeRowSizeLimit( + IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit()); } public Response executeQueryInternal( @@ -103,7 +105,8 @@ public Response executeQueryInternal( QueryDataSetHandler.fillQueryDataSet( queryExecution, statement, - sql.getRowLimit() == null ? defaultQueryRowLimit : sql.getRowLimit()); + QueryRowLimitUtils.resolveActualRowSizeLimit( + sql.getRowLimit(), defaultQueryRowLimit)); if (queryExecution.getQueryType() == QueryType.READ_WRITE) { return responseGenerateHelper(result); } diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandler.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandler.java index 3918a07dd52d0..4cee0926e7753 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandler.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandler.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.model.ShowLoadedModelsStatement; import org.apache.iotdb.db.queryengine.plan.statement.metadata.model.ShowModelsStatement; import org.apache.iotdb.db.queryengine.plan.statement.sys.AuthorStatement; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.v1.model.ExecutionStatus; import org.apache.iotdb.rpc.TSStatusCode; @@ -51,7 +52,7 @@ public class QueryDataSetHandler { private QueryDataSetHandler() {} /** - * @param actualRowSizeLimit max number of rows to return. no limit when actualRowSizeLimit <= 0. + * @param actualRowSizeLimit max number of rows to return. */ public static Response fillQueryDataSet( IQueryExecution queryExecution, Statement statement, int actualRowSizeLimit) @@ -152,7 +153,6 @@ private static Response fillQueryDataSetWithTimestamps( final long timePrecision) throws IoTDBException { int fetched = 0; - int columnNum = queryExecution.getOutputValueColumnCount(); DatasetHeader header = queryExecution.getDatasetHeader(); List resultColumns = header.getRespColumns(); @@ -164,17 +164,6 @@ private static Response fillQueryDataSetWithTimestamps( } while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new org.apache.iotdb.rest.protocol.model.ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { if (fetched == 0) { @@ -186,6 +175,9 @@ private static Response fillQueryDataSetWithTimestamps( } TsBlock tsBlock = optionalTsBlock.get(); int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } // time column for (int i = 0; i < currentCount; i++) { targetDataSet.addTimestampsItem( @@ -197,17 +189,14 @@ private static Response fillQueryDataSetWithTimestamps( Column column = tsBlock.getColumn(headerMap.get(resultColumns.get(k))); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(dataTypes, k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } return Response.ok().entity(targetDataSet).build(); } @@ -222,17 +211,6 @@ private static Response fillQueryDataSetWithoutTimestamps( int fetched = 0; int columnNum = queryExecution.getOutputValueColumnCount(); while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new org.apache.iotdb.rest.protocol.model.ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent()) { if (fetched == 0) { @@ -247,21 +225,21 @@ private static Response fillQueryDataSetWithoutTimestamps( targetDataSet.setValues(new ArrayList<>()); return Response.ok().entity(targetDataSet).build(); } + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } for (int k = 0; k < columnNum; k++) { Column column = tsBlock.getColumn(targetDataSetIndexToSourceDataSetIndex[k]); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(dataTypes, k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } return Response.ok().entity(targetDataSet).build(); } @@ -291,44 +269,60 @@ private static void addTypedValueToTarget( } public static Response fillGrafanaVariablesResult( - IQueryExecution queryExecution, Statement statement) throws IoTDBException { + IQueryExecution queryExecution, Statement statement, int actualRowSizeLimit) + throws IoTDBException { List results = new ArrayList<>(); - Optional optionalTsBlock = queryExecution.getBatchResult(); - if (!optionalTsBlock.isPresent()) { - return Response.ok().entity(results).build(); - } - TsBlock tsBlock = optionalTsBlock.get(); - int currentCount = tsBlock.getPositionCount(); - Column column = tsBlock.getColumn(0); + int fetched = 0; + while (true) { + Optional optionalTsBlock = queryExecution.getBatchResult(); + if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { + return Response.ok().entity(results).build(); + } + TsBlock tsBlock = optionalTsBlock.get(); + int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } + Column column = tsBlock.getColumn(0); - for (int i = 0; i < currentCount; i++) { - String nodePaths = column.getObject(i).toString(); - if (statement instanceof ShowChildPathsStatement) { - String[] nodeSubPath = nodePaths.split("\\."); - results.add(nodeSubPath[nodeSubPath.length - 1]); - } else { - results.add(nodePaths); + for (int i = 0; i < currentCount; i++) { + String nodePaths = column.getObject(i).toString(); + if (statement instanceof ShowChildPathsStatement) { + String[] nodeSubPath = nodePaths.split("\\."); + results.add(nodeSubPath[nodeSubPath.length - 1]); + } else { + results.add(nodePaths); + } } + fetched += currentCount; } - return Response.ok().entity(results).build(); } - public static Response fillGrafanaNodesResult(IQueryExecution queryExecution) - throws IoTDBException { + public static Response fillGrafanaNodesResult( + IQueryExecution queryExecution, int actualRowSizeLimit) throws IoTDBException { List nodes = new ArrayList<>(); - Optional optionalTsBlock = queryExecution.getBatchResult(); - if (!optionalTsBlock.isPresent()) { + if (queryExecution == null) { return Response.ok().entity(nodes).build(); } - TsBlock tsBlock = optionalTsBlock.get(); - int currentCount = tsBlock.getPositionCount(); - Column column = tsBlock.getColumn(0); + int fetched = 0; + while (true) { + Optional optionalTsBlock = queryExecution.getBatchResult(); + if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { + return Response.ok().entity(nodes).build(); + } + TsBlock tsBlock = optionalTsBlock.get(); + int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } + Column column = tsBlock.getColumn(0); - for (int i = 0; i < currentCount; i++) { - String nodePaths = column.getObject(i).toString(); - String[] nodeSubPath = nodePaths.split("\\."); - nodes.add(nodeSubPath[nodeSubPath.length - 1]); + for (int i = 0; i < currentCount; i++) { + String nodePaths = column.getObject(i).toString(); + String[] nodeSubPath = nodePaths.split("\\."); + nodes.add(nodeSubPath[nodeSubPath.length - 1]); + } + fetched += currentCount; } - return Response.ok().entity(nodes).build(); } } diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/GrafanaApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/GrafanaApiServiceImpl.java index 19b8ab8b38ec4..dcc17c43a59d9 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/GrafanaApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/GrafanaApiServiceImpl.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor; import org.apache.iotdb.db.protocol.session.IClientSession; import org.apache.iotdb.db.protocol.session.SessionManager; import org.apache.iotdb.db.queryengine.plan.Coordinator; @@ -36,6 +37,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowStatement; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.iotdb.rest.protocol.handler.AuthorizationHandler; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.v1.GrafanaApiService; import org.apache.iotdb.rest.protocol.v1.handler.ExceptionHandler; import org.apache.iotdb.rest.protocol.v1.handler.QueryDataSetHandler; @@ -68,11 +70,15 @@ public class GrafanaApiServiceImpl extends GrafanaApiService { private final AuthorizationHandler authorizationHandler; private final long timePrecision; // the default timestamp precision is ms + private final int defaultQueryRowLimit; public GrafanaApiServiceImpl() { partitionFetcher = ClusterPartitionFetcher.getInstance(); schemaFetcher = ClusterSchemaFetcher.getInstance(); authorizationHandler = new AuthorizationHandler(); + defaultQueryRowLimit = + QueryRowLimitUtils.normalizeRowSizeLimit( + IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit()); switch (CommonDescriptor.getInstance().getConfig().getTimestampPrecision()) { case "ns": @@ -132,7 +138,8 @@ public Response variables(SQL sql, SecurityContext securityContext) { } IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId); try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { - return QueryDataSetHandler.fillGrafanaVariablesResult(queryExecution, statement); + return QueryDataSetHandler.fillGrafanaVariablesResult( + queryExecution, statement, defaultQueryRowLimit); } } catch (Exception e) { return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build(); @@ -204,9 +211,11 @@ public Response expression(ExpressionRequest expressionRequest, SecurityContext try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { if (((QueryStatement) statement).isAggregationQuery() && !((QueryStatement) statement).isGroupByTime()) { - return QueryDataSetHandler.fillAggregationPlanDataSet(queryExecution, 0); + return QueryDataSetHandler.fillAggregationPlanDataSet( + queryExecution, defaultQueryRowLimit); } else { - return QueryDataSetHandler.fillDataSetWithTimestamps(queryExecution, 0, timePrecision); + return QueryDataSetHandler.fillDataSetWithTimestamps( + queryExecution, defaultQueryRowLimit, timePrecision); } } } catch (Exception e) { @@ -269,10 +278,10 @@ public Response node(List requestBody, SecurityContext securityContext) IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId); try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { - return QueryDataSetHandler.fillGrafanaNodesResult(queryExecution); + return QueryDataSetHandler.fillGrafanaNodesResult(queryExecution, defaultQueryRowLimit); } } else { - return QueryDataSetHandler.fillGrafanaNodesResult(null); + return QueryDataSetHandler.fillGrafanaNodesResult(null, defaultQueryRowLimit); } } catch (Exception e) { return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build(); diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/RestApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/RestApiServiceImpl.java index b5b57ac10e7be..4939e10a41246 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/RestApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v1/impl/RestApiServiceImpl.java @@ -39,6 +39,7 @@ import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.iotdb.rest.protocol.handler.AuthorizationHandler; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.utils.InsertTabletSortDataUtils; import org.apache.iotdb.rest.protocol.v1.RestApiService; import org.apache.iotdb.rest.protocol.v1.handler.ExceptionHandler; @@ -70,14 +71,15 @@ public class RestApiServiceImpl extends RestApiService { private final ISchemaFetcher schemaFetcher; private final AuthorizationHandler authorizationHandler; - private final Integer defaultQueryRowLimit; + private final int defaultQueryRowLimit; public RestApiServiceImpl() { partitionFetcher = ClusterPartitionFetcher.getInstance(); schemaFetcher = ClusterSchemaFetcher.getInstance(); authorizationHandler = new AuthorizationHandler(); defaultQueryRowLimit = - IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit(); + QueryRowLimitUtils.normalizeRowSizeLimit( + IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit()); } @Override @@ -231,7 +233,7 @@ public Response executeQueryStatement(SQL sql, SecurityContext securityContext) return QueryDataSetHandler.fillQueryDataSet( queryExecution, statement, - sql.getRowLimit() == null ? defaultQueryRowLimit : sql.getRowLimit()); + QueryRowLimitUtils.resolveActualRowSizeLimit(sql.getRowLimit(), defaultQueryRowLimit)); } } catch (Exception e) { finish = true; diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/QueryDataSetHandler.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/QueryDataSetHandler.java index b188454e17977..34d03eac6ecd6 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/QueryDataSetHandler.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/QueryDataSetHandler.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.model.ShowLoadedModelsStatement; import org.apache.iotdb.db.queryengine.plan.statement.metadata.model.ShowModelsStatement; import org.apache.iotdb.db.queryengine.plan.statement.sys.AuthorStatement; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.model.ExecutionStatus; import org.apache.iotdb.rpc.TSStatusCode; @@ -51,7 +52,7 @@ public class QueryDataSetHandler { private QueryDataSetHandler() {} /** - * @param actualRowSizeLimit max number of rows to return. no limit when actualRowSizeLimit <= 0. + * @param actualRowSizeLimit max number of rows to return. */ public static Response fillQueryDataSet( IQueryExecution queryExecution, Statement statement, int actualRowSizeLimit) @@ -145,7 +146,6 @@ private static Response fillQueryDataSetWithTimestamps( final long timePrecision) throws IoTDBException { int fetched = 0; - int columnNum = queryExecution.getOutputValueColumnCount(); DatasetHeader header = queryExecution.getDatasetHeader(); List resultColumns = header.getRespColumns(); @@ -158,17 +158,6 @@ private static Response fillQueryDataSetWithTimestamps( } while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { if (fetched == 0) { @@ -180,6 +169,9 @@ private static Response fillQueryDataSetWithTimestamps( } TsBlock tsBlock = optionalTsBlock.get(); int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } // time column for (int i = 0; i < currentCount; i++) { targetDataSet.addTimestampsItem( @@ -191,17 +183,14 @@ private static Response fillQueryDataSetWithTimestamps( Column column = tsBlock.getColumn(headerMap.get(resultColumns.get(k))); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(targetDataSet.getDataTypes(), k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } return Response.ok().entity(targetDataSet).build(); } @@ -216,17 +205,6 @@ private static Response fillQueryDataSetWithoutTimestamps( int columnNum = queryExecution.getOutputValueColumnCount(); List dataTypes = targetDataSet.getDataTypes(); while (true) { - if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) { - return Response.ok() - .entity( - new ExecutionStatus() - .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode()) - .message( - String.format( - "Dataset row size exceeded the given max row size (%d)", - actualRowSizeLimit))) - .build(); - } Optional optionalTsBlock = queryExecution.getBatchResult(); if (!optionalTsBlock.isPresent()) { if (fetched == 0) { @@ -241,21 +219,21 @@ private static Response fillQueryDataSetWithoutTimestamps( targetDataSet.setValues(new ArrayList<>()); return Response.ok().entity(targetDataSet).build(); } + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } for (int k = 0; k < columnNum; k++) { Column column = tsBlock.getColumn(targetDataSetIndexToSourceDataSetIndex[k]); List targetDataSetColumn = targetDataSet.getValues().get(k); for (int i = 0; i < currentCount; i++) { - fetched++; if (column.isNull(i)) { targetDataSetColumn.add(null); } else { addTypedValueToTarget(dataTypes, k, i, targetDataSetColumn, column); } } - if (k != columnNum - 1) { - fetched -= currentCount; - } } + fetched += currentCount; } return Response.ok().entity(targetDataSet).build(); } @@ -285,44 +263,60 @@ private static void addTypedValueToTarget( } public static Response fillGrafanaVariablesResult( - IQueryExecution queryExecution, Statement statement) throws IoTDBException { + IQueryExecution queryExecution, Statement statement, int actualRowSizeLimit) + throws IoTDBException { List results = new ArrayList<>(); - Optional optionalTsBlock = queryExecution.getBatchResult(); - if (!optionalTsBlock.isPresent()) { - return Response.ok().entity(results).build(); - } - TsBlock tsBlock = optionalTsBlock.get(); - int currentCount = tsBlock.getPositionCount(); - Column column = tsBlock.getColumn(0); + int fetched = 0; + while (true) { + Optional optionalTsBlock = queryExecution.getBatchResult(); + if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { + return Response.ok().entity(results).build(); + } + TsBlock tsBlock = optionalTsBlock.get(); + int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } + Column column = tsBlock.getColumn(0); - for (int i = 0; i < currentCount; i++) { - String nodePaths = column.getObject(i).toString(); - if (statement instanceof ShowChildPathsStatement) { - String[] nodeSubPath = nodePaths.split("\\."); - results.add(nodeSubPath[nodeSubPath.length - 1]); - } else { - results.add(nodePaths); + for (int i = 0; i < currentCount; i++) { + String nodePaths = column.getObject(i).toString(); + if (statement instanceof ShowChildPathsStatement) { + String[] nodeSubPath = nodePaths.split("\\."); + results.add(nodeSubPath[nodeSubPath.length - 1]); + } else { + results.add(nodePaths); + } } + fetched += currentCount; } - return Response.ok().entity(results).build(); } - public static Response fillGrafanaNodesResult(IQueryExecution queryExecution) - throws IoTDBException { + public static Response fillGrafanaNodesResult( + IQueryExecution queryExecution, int actualRowSizeLimit) throws IoTDBException { List nodes = new ArrayList<>(); - Optional optionalTsBlock = queryExecution.getBatchResult(); - if (!optionalTsBlock.isPresent()) { + if (queryExecution == null) { return Response.ok().entity(nodes).build(); } - TsBlock tsBlock = optionalTsBlock.get(); - int currentCount = tsBlock.getPositionCount(); - Column column = tsBlock.getColumn(0); + int fetched = 0; + while (true) { + Optional optionalTsBlock = queryExecution.getBatchResult(); + if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) { + return Response.ok().entity(nodes).build(); + } + TsBlock tsBlock = optionalTsBlock.get(); + int currentCount = tsBlock.getPositionCount(); + if (QueryRowLimitUtils.exceedsLimit(fetched, currentCount, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } + Column column = tsBlock.getColumn(0); - for (int i = 0; i < currentCount; i++) { - String nodePaths = column.getObject(i).toString(); - String[] nodeSubPath = nodePaths.split("\\."); - nodes.add(nodeSubPath[nodeSubPath.length - 1]); + for (int i = 0; i < currentCount; i++) { + String nodePaths = column.getObject(i).toString(); + String[] nodeSubPath = nodePaths.split("\\."); + nodes.add(nodeSubPath[nodeSubPath.length - 1]); + } + fetched += currentCount; } - return Response.ok().entity(nodes).build(); } } diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/GrafanaApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/GrafanaApiServiceImpl.java index e8701f89347a5..32834a5bced0d 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/GrafanaApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/GrafanaApiServiceImpl.java @@ -21,6 +21,7 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.conf.IoTDBConfig; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.conf.rest.IoTDBRestServiceDescriptor; import org.apache.iotdb.db.protocol.session.IClientSession; import org.apache.iotdb.db.protocol.session.SessionManager; import org.apache.iotdb.db.queryengine.plan.Coordinator; @@ -36,6 +37,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.metadata.ShowStatement; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.iotdb.rest.protocol.handler.AuthorizationHandler; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.v2.GrafanaApiService; import org.apache.iotdb.rest.protocol.v2.handler.ExceptionHandler; import org.apache.iotdb.rest.protocol.v2.handler.QueryDataSetHandler; @@ -68,11 +70,15 @@ public class GrafanaApiServiceImpl extends GrafanaApiService { private final AuthorizationHandler authorizationHandler; private final long timePrecision; // the default timestamp precision is ms + private final int defaultQueryRowLimit; public GrafanaApiServiceImpl() { partitionFetcher = ClusterPartitionFetcher.getInstance(); schemaFetcher = ClusterSchemaFetcher.getInstance(); authorizationHandler = new AuthorizationHandler(); + defaultQueryRowLimit = + QueryRowLimitUtils.normalizeRowSizeLimit( + IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit()); switch (CommonDescriptor.getInstance().getConfig().getTimestampPrecision()) { case "ns": @@ -132,7 +138,8 @@ public Response variables(SQL sql, SecurityContext securityContext) { } IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId); try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { - return QueryDataSetHandler.fillGrafanaVariablesResult(queryExecution, statement); + return QueryDataSetHandler.fillGrafanaVariablesResult( + queryExecution, statement, defaultQueryRowLimit); } } catch (Exception e) { return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build(); @@ -204,9 +211,11 @@ public Response expression(ExpressionRequest expressionRequest, SecurityContext try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { if (((QueryStatement) statement).isAggregationQuery() && !((QueryStatement) statement).isGroupByTime()) { - return QueryDataSetHandler.fillAggregationPlanDataSet(queryExecution, 0); + return QueryDataSetHandler.fillAggregationPlanDataSet( + queryExecution, defaultQueryRowLimit); } else { - return QueryDataSetHandler.fillDataSetWithTimestamps(queryExecution, 0, timePrecision); + return QueryDataSetHandler.fillDataSetWithTimestamps( + queryExecution, defaultQueryRowLimit, timePrecision); } } } catch (Exception e) { @@ -269,10 +278,10 @@ public Response node(List requestBody, SecurityContext securityContext) IQueryExecution queryExecution = COORDINATOR.getQueryExecution(queryId); try (SetThreadName threadName = new SetThreadName(result.queryId.getId())) { - return QueryDataSetHandler.fillGrafanaNodesResult(queryExecution); + return QueryDataSetHandler.fillGrafanaNodesResult(queryExecution, defaultQueryRowLimit); } } else { - return QueryDataSetHandler.fillGrafanaNodesResult(null); + return QueryDataSetHandler.fillGrafanaNodesResult(null, defaultQueryRowLimit); } } catch (Exception e) { return Response.ok().entity(ExceptionHandler.tryCatchException(e)).build(); diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java index 6e2cef45b7f23..c1dfd3e35d2e5 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java @@ -47,6 +47,7 @@ import org.apache.iotdb.db.utils.CommonUtils; import org.apache.iotdb.db.utils.SetThreadName; import org.apache.iotdb.rest.protocol.handler.AuthorizationHandler; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.model.ExecutionStatus; import org.apache.iotdb.rest.protocol.utils.InsertTabletSortDataUtils; import org.apache.iotdb.rest.protocol.v2.NotFoundException; @@ -99,14 +100,15 @@ public class RestApiServiceImpl extends RestApiService { private final ISchemaFetcher schemaFetcher; private final AuthorizationHandler authorizationHandler; - private final Integer defaultQueryRowLimit; + private final int defaultQueryRowLimit; public RestApiServiceImpl() { partitionFetcher = ClusterPartitionFetcher.getInstance(); schemaFetcher = ClusterSchemaFetcher.getInstance(); authorizationHandler = new AuthorizationHandler(); defaultQueryRowLimit = - IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit(); + QueryRowLimitUtils.normalizeRowSizeLimit( + IoTDBRestServiceDescriptor.getInstance().getConfig().getRestQueryDefaultRowSizeLimit()); } @Override @@ -187,6 +189,7 @@ public Response executeFastLastQueryStatement( List timeseries = new ArrayList<>(); List valueList = new ArrayList<>(); List dataTypeList = new ArrayList<>(); + int fetched = 0; for (final Map.Entry>>> result : resultMap.entrySet()) { @@ -201,10 +204,14 @@ public Response executeFastLastQueryStatement( || tvPair.getValue() == null) { continue; } + if (QueryRowLimitUtils.exceedsLimit(fetched, 1, defaultQueryRowLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(defaultQueryRowLimit); + } valueList.add(tvPair.getValue().getStringValue()); dataTypeList.add(tvPair.getValue().getDataType().name()); targetDataSet.addTimestampsItem(tvPair.getTimestamp()); timeseries.add(deviceWithSeparator + measurementLastEntry.getKey()); + fetched++; } } } @@ -398,7 +405,7 @@ public Response executeQueryStatement(SQL sql, SecurityContext securityContext) return QueryDataSetHandler.fillQueryDataSet( queryExecution, statement, - sql.getRowLimit() == null ? defaultQueryRowLimit : sql.getRowLimit()); + QueryRowLimitUtils.resolveActualRowSizeLimit(sql.getRowLimit(), defaultQueryRowLimit)); } } catch (Exception e) { finish = true; diff --git a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java new file mode 100644 index 0000000000000..02bdaf1c7fec9 --- /dev/null +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java @@ -0,0 +1,52 @@ +/* + * 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.rest.protocol.handler; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class QueryRowLimitUtilsTest { + + @Test + public void resolveActualRowSizeLimitShouldUseConfiguredLimitAsHardLimit() { + assertEquals(10, QueryRowLimitUtils.resolveActualRowSizeLimit(null, 10)); + assertEquals(10, QueryRowLimitUtils.resolveActualRowSizeLimit(100, 10)); + assertEquals(5, QueryRowLimitUtils.resolveActualRowSizeLimit(5, 10)); + } + + @Test + public void resolveActualRowSizeLimitShouldKeepHardLimitPositive() { + assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(null, 0)); + assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(100, 0)); + assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(null, -1)); + } + + @Test + public void exceedsLimitShouldRejectOnlyRowsBeyondTheHardLimit() { + assertFalse(QueryRowLimitUtils.exceedsLimit(0, 2, 2)); + assertTrue(QueryRowLimitUtils.exceedsLimit(2, 1, 2)); + assertTrue(QueryRowLimitUtils.exceedsLimit(0, 2, 1)); + assertFalse(QueryRowLimitUtils.exceedsLimit(0, 0, 1)); + assertTrue(QueryRowLimitUtils.exceedsLimit(0, 2, 0)); + } +} diff --git a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandlerTest.java b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandlerTest.java new file mode 100644 index 0000000000000..36a2ec1292e17 --- /dev/null +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v1/handler/QueryDataSetHandlerTest.java @@ -0,0 +1,222 @@ +/* + * 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.rest.protocol.v1.handler; + +import org.apache.iotdb.commons.exception.IoTDBException; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; +import org.apache.iotdb.commons.schema.column.ColumnHeader; +import org.apache.iotdb.db.queryengine.common.header.DatasetHeader; +import org.apache.iotdb.db.queryengine.plan.analyze.QueryType; +import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult; +import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution; +import org.apache.iotdb.rest.protocol.model.ExecutionStatus; +import org.apache.iotdb.rest.protocol.v1.model.QueryDataSet; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.common.block.TsBlock; +import org.apache.tsfile.read.common.block.TsBlockBuilder; +import org.junit.Test; + +import jakarta.ws.rs.core.Response; + +import java.nio.ByteBuffer; +import java.util.ArrayDeque; +import java.util.Arrays; +import java.util.Collections; +import java.util.Optional; +import java.util.Queue; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class QueryDataSetHandlerTest { + + @Test + public void fillDataSetWithTimestampsShouldAllowResultAtExactLimit() throws Exception { + Response response = + QueryDataSetHandler.fillDataSetWithTimestamps( + new TestQueryExecution(newTsBlock(1L, 11L, 2L, 22L)), 2, 1); + + assertTrue(response.getEntity() instanceof QueryDataSet); + QueryDataSet dataSet = (QueryDataSet) response.getEntity(); + assertEquals(Arrays.asList(1L, 2L), dataSet.getTimestamps()); + assertEquals(Collections.singletonList(Arrays.asList(11L, 22L)), dataSet.getValues()); + } + + @Test + public void fillDataSetWithTimestampsShouldRejectRowsBeyondLimit() throws Exception { + Response response = + QueryDataSetHandler.fillDataSetWithTimestamps( + new TestQueryExecution(newTsBlock(1L, 11L, 2L, 22L)), 1, 1); + + assertTrue(response.getEntity() instanceof ExecutionStatus); + ExecutionStatus status = (ExecutionStatus) response.getEntity(); + assertEquals(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode(), status.getCode().intValue()); + } + + private static TsBlock newTsBlock(long... timeAndValues) { + TsBlockBuilder builder = new TsBlockBuilder(Collections.singletonList(TSDataType.INT64)); + for (int i = 0; i < timeAndValues.length; i += 2) { + builder.getTimeColumnBuilder().writeLong(timeAndValues[i]); + builder.getColumnBuilder(0).writeLong(timeAndValues[i + 1]); + builder.declarePosition(); + } + return builder.build(); + } + + private static final class TestQueryExecution implements IQueryExecution { + + private static final String COLUMN_NAME = "root.sg.d1.s1"; + + private final Queue> batches = new ArrayDeque<>(); + private final DatasetHeader datasetHeader; + + private TestQueryExecution(TsBlock... tsBlocks) { + for (TsBlock tsBlock : tsBlocks) { + batches.add(Optional.of(tsBlock)); + } + batches.add(Optional.empty()); + + datasetHeader = + new DatasetHeader( + Collections.singletonList(new ColumnHeader(COLUMN_NAME, TSDataType.INT64)), false); + datasetHeader.setTreeColumnToTsBlockIndexMap(Collections.singletonList(COLUMN_NAME)); + } + + @Override + public void start() {} + + @Override + public void stop(Throwable t) {} + + @Override + public void stopAndCleanup(Throwable t) {} + + @Override + public void cancel() {} + + @Override + public ExecutionResult getStatus() { + return null; + } + + @Override + public Optional getBatchResult() throws IoTDBException { + return batches.remove(); + } + + @Override + public Optional getByteBufferBatchResult() { + return Optional.empty(); + } + + @Override + public boolean hasNextResult() { + return !batches.isEmpty(); + } + + @Override + public int getOutputValueColumnCount() { + return 1; + } + + @Override + public DatasetHeader getDatasetHeader() { + return datasetHeader; + } + + @Override + public QueryType getQueryType() { + return QueryType.READ; + } + + @Override + public boolean isQuery() { + return true; + } + + @Override + public boolean isUserQuery() { + return true; + } + + @Override + public String getQueryId() { + return null; + } + + @Override + public long getStartExecutionTime() { + return 0; + } + + @Override + public void recordExecutionTime(long executionTime) {} + + @Override + public void updateCurrentRpcStartTime(long startTime) {} + + @Override + public boolean isActive() { + return false; + } + + @Override + public long getTotalExecutionTime() { + return 0; + } + + @Override + public long getTimeout() { + return 0; + } + + @Override + public Optional getExecuteSQL() { + return Optional.empty(); + } + + @Override + public String getStatementType() { + return null; + } + + @Override + public SqlDialect getSQLDialect() { + return SqlDialect.TREE; + } + + @Override + public String getUser() { + return null; + } + + @Override + public String getClientHostname() { + return null; + } + + @Override + public boolean isDebug() { + return false; + } + } +} diff --git a/iotdb-client/session/src/test/resources/iotdb-common.properties b/iotdb-client/session/src/test/resources/iotdb-common.properties index 2ec58b5db3255..5c2150a3789d9 100644 --- a/iotdb-client/session/src/test/resources/iotdb-common.properties +++ b/iotdb-client/session/src/test/resources/iotdb-common.properties @@ -27,7 +27,8 @@ enable_rest_service=true # the binding port of the REST service # rest_service_port=18080 -# the default row limit to a REST query response when the rowSize parameter is not given in request +# The maximum row limit for REST and Grafana query responses. +# The request rowLimit/row_limit value cannot exceed this limit. # rest_query_default_row_size_limit=10000 # Whether to display rest service interface information through OpenAPI. eg: http://ip:port/openapi.json diff --git a/iotdb-core/datanode/src/test/resources/iotdb-common.properties b/iotdb-core/datanode/src/test/resources/iotdb-common.properties index b753b0519ee9d..d22cfaa637240 100644 --- a/iotdb-core/datanode/src/test/resources/iotdb-common.properties +++ b/iotdb-core/datanode/src/test/resources/iotdb-common.properties @@ -30,7 +30,8 @@ enable_rest_service=true # Whether to display rest service interface information through OpenAPI. eg: http://ip:port/openapi.json # enable_swagger=false -# the default row limit to a REST query response when the rowSize parameter is not given in request +# The maximum row limit for REST and Grafana query responses. +# The request rowLimit/row_limit value cannot exceed this limit. # rest_query_default_row_size_limit=10000 # is SSL enabled diff --git a/iotdb-core/datanode/src/test/resources/iotdb-system.properties b/iotdb-core/datanode/src/test/resources/iotdb-system.properties index a7a2a08bd1fe4..4cde9826f0793 100644 --- a/iotdb-core/datanode/src/test/resources/iotdb-system.properties +++ b/iotdb-core/datanode/src/test/resources/iotdb-system.properties @@ -48,7 +48,8 @@ enable_rest_service=true # Whether to display rest service interface information through OpenAPI. eg: http://ip:port/openapi.json # enable_swagger=false -# the default row limit to a REST query response when the rowSize parameter is not given in request +# The maximum row limit for REST and Grafana query responses. +# The request rowLimit/row_limit value cannot exceed this limit. # rest_query_default_row_size_limit=10000 # is SSL enabled 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 fdf7994325540..91e84e89a0ece 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 @@ -607,7 +607,8 @@ rest_service_port=18080 # Datatype: boolean enable_swagger=false -# the default row limit to a REST query response when the rowSize parameter is not given in request +# The maximum row limit for REST and Grafana query responses. +# The request rowLimit/row_limit value cannot exceed this limit. # effectiveMode: restart # Datatype: int rest_query_default_row_size_limit=10000 From 295ef2fa6f6e3bd0fcb7497dc4b59d0e1ff8a78a Mon Sep 17 00:00:00 2001 From: JackieTien97 Date: Wed, 8 Jul 2026 15:36:26 +0800 Subject: [PATCH 2/2] Remap non-positive REST row limit config to default and add regression tests A non-positive rest_query_default_row_size_limit used to mean "unlimited"; the hard-cap handling clamped it down to 1, silently capping every REST response to a single row. Remap non-positive values to the built-in default (10000) so existing deployments are not degraded, while still guaranteeing a finite cap. Add regression coverage: - QueryRowLimitUtilsTest: an oversized caller-provided rowLimit is clamped to the configured hard limit. - FastLastHandlerTest: the fastLastQuery cache-hit path caps the number of materialized entries. The cache-hit loop is extracted into FastLastHandler.fillLastValueDataSet so it is unit-testable. --- .../protocol/handler/QueryRowLimitUtils.java | 9 +- .../protocol/v2/handler/FastLastHandler.java | 65 +++++++++- .../protocol/v2/impl/RestApiServiceImpl.java | 45 +------ .../handler/QueryRowLimitUtilsTest.java | 17 ++- .../v2/handler/FastLastHandlerTest.java | 119 ++++++++++++++++++ .../conf/iotdb-system.properties.template | 1 + 6 files changed, 204 insertions(+), 52 deletions(-) create mode 100644 external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandlerTest.java diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java index 4d8bd08b4f004..583b1386b4fe3 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtils.java @@ -27,7 +27,12 @@ public final class QueryRowLimitUtils { - private static final int MIN_ROW_SIZE_LIMIT = 1; + /** + * Fallback used when {@code rest_query_default_row_size_limit} is missing or non-positive. + * Matches the built-in default in {@code IoTDBRestServiceConfig}; a non-positive configured value + * used to mean "unlimited" and is now treated as this cap instead of being clamped down to 1. + */ + private static final int DEFAULT_ROW_SIZE_LIMIT = 10000; private QueryRowLimitUtils() {} @@ -41,7 +46,7 @@ public static int resolveActualRowSizeLimit( } public static int normalizeRowSizeLimit(int rowSizeLimit) { - return Math.max(MIN_ROW_SIZE_LIMIT, rowSizeLimit); + return rowSizeLimit > 0 ? rowSizeLimit : DEFAULT_ROW_SIZE_LIMIT; } public static boolean exceedsLimit( diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandler.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandler.java index 5bb5e1274ae2e..ccc0637346381 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandler.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandler.java @@ -19,15 +19,27 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.db.protocol.session.IClientSession; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableId; +import org.apache.iotdb.rest.protocol.handler.QueryRowLimitUtils; import org.apache.iotdb.rest.protocol.v2.model.ExecutionStatus; import org.apache.iotdb.rest.protocol.v2.model.PrefixPathList; +import org.apache.iotdb.rest.protocol.v2.model.QueryDataSet; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSLastDataQueryReq; +import org.apache.tsfile.common.constant.TsFileConstant; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.utils.Pair; + import jakarta.ws.rs.core.Response; import java.util.ArrayList; import java.util.Collections; +import java.util.List; +import java.util.Map; public class FastLastHandler { @@ -57,8 +69,7 @@ public static Response buildExecutionStatusResponse(TSStatus status) { .build(); } - public static void setupTargetDataSet( - org.apache.iotdb.rest.protocol.v2.model.QueryDataSet dataSet) { + public static void setupTargetDataSet(QueryDataSet dataSet) { dataSet.addExpressionsItem("Timeseries"); dataSet.addExpressionsItem("Value"); dataSet.addExpressionsItem("DataType"); @@ -68,4 +79,54 @@ public static void setupTargetDataSet( dataSet.setValues(new ArrayList<>()); dataSet.setTimestamps(new ArrayList<>()); } + + /** + * Builds the fastLastQuery response directly from cached last values. The number of materialized + * entries is capped by {@code actualRowSizeLimit}; once exceeded an "exceeded max row size" + * response is returned instead of materializing the whole cache into heap. + * + * @param resultMap last values keyed by table/device/measurement, as filled by the schema cache + * @param actualRowSizeLimit hard cap on the number of returned rows + */ + public static Response fillLastValueDataSet( + Map>>> resultMap, + int actualRowSizeLimit) { + QueryDataSet targetDataSet = new QueryDataSet(); + setupTargetDataSet(targetDataSet); + List timeseries = new ArrayList<>(); + List valueList = new ArrayList<>(); + List dataTypeList = new ArrayList<>(); + int fetched = 0; + + for (final Map.Entry>>> + result : resultMap.entrySet()) { + for (final Map.Entry>> + device2MeasurementLastEntry : result.getValue().entrySet()) { + final String deviceWithSeparator = + device2MeasurementLastEntry.getKey().toString() + TsFileConstant.PATH_SEPARATOR; + for (final Map.Entry> measurementLastEntry : + device2MeasurementLastEntry.getValue().entrySet()) { + final TimeValuePair tvPair = measurementLastEntry.getValue().getRight(); + if (tvPair == TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN + || tvPair.getValue() == null) { + continue; + } + if (QueryRowLimitUtils.exceedsLimit(fetched, 1, actualRowSizeLimit)) { + return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(actualRowSizeLimit); + } + valueList.add(tvPair.getValue().getStringValue()); + dataTypeList.add(tvPair.getValue().getDataType().name()); + targetDataSet.addTimestampsItem(tvPair.getTimestamp()); + timeseries.add(deviceWithSeparator + measurementLastEntry.getKey()); + fetched++; + } + } + } + if (!timeseries.isEmpty()) { + targetDataSet.addValuesItem(timeseries); + targetDataSet.addValuesItem(valueList); + targetDataSet.addValuesItem(dataTypeList); + } + return Response.ok().entity(targetDataSet).build(); + } } diff --git a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java index c1dfd3e35d2e5..cfe572ce9e046 100644 --- a/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java +++ b/external-service-impl/rest/src/main/java/org/apache/iotdb/rest/protocol/v2/impl/RestApiServiceImpl.java @@ -34,7 +34,6 @@ import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult; import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution; import org.apache.iotdb.db.queryengine.plan.parser.StatementGenerator; -import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache; import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableId; import org.apache.iotdb.db.queryengine.plan.statement.Statement; @@ -61,12 +60,10 @@ import org.apache.iotdb.rest.protocol.v2.model.InsertRecordsRequest; import org.apache.iotdb.rest.protocol.v2.model.InsertTabletRequest; import org.apache.iotdb.rest.protocol.v2.model.PrefixPathList; -import org.apache.iotdb.rest.protocol.v2.model.QueryDataSet; import org.apache.iotdb.rest.protocol.v2.model.SQL; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.service.rpc.thrift.TSLastDataQueryReq; -import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.read.TimeValuePair; @@ -76,7 +73,6 @@ import jakarta.ws.rs.core.SecurityContext; import java.time.ZoneId; -import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -182,46 +178,9 @@ public Response executeFastLastQueryStatement( } } - // Cache hit: build response directly - QueryDataSet targetDataSet = new QueryDataSet(); - - FastLastHandler.setupTargetDataSet(targetDataSet); - List timeseries = new ArrayList<>(); - List valueList = new ArrayList<>(); - List dataTypeList = new ArrayList<>(); - int fetched = 0; - - for (final Map.Entry>>> - result : resultMap.entrySet()) { - for (final Map.Entry>> - device2MeasurementLastEntry : result.getValue().entrySet()) { - final String deviceWithSeparator = - device2MeasurementLastEntry.getKey().toString() + TsFileConstant.PATH_SEPARATOR; - for (final Map.Entry> measurementLastEntry : - device2MeasurementLastEntry.getValue().entrySet()) { - final TimeValuePair tvPair = measurementLastEntry.getValue().getRight(); - if (tvPair == TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN - || tvPair.getValue() == null) { - continue; - } - if (QueryRowLimitUtils.exceedsLimit(fetched, 1, defaultQueryRowLimit)) { - return QueryRowLimitUtils.buildRowSizeLimitExceededResponse(defaultQueryRowLimit); - } - valueList.add(tvPair.getValue().getStringValue()); - dataTypeList.add(tvPair.getValue().getDataType().name()); - targetDataSet.addTimestampsItem(tvPair.getTimestamp()); - timeseries.add(deviceWithSeparator + measurementLastEntry.getKey()); - fetched++; - } - } - } - if (!timeseries.isEmpty()) { - targetDataSet.addValuesItem(timeseries); - targetDataSet.addValuesItem(valueList); - targetDataSet.addValuesItem(dataTypeList); - } + // Cache hit: build response directly (capped by defaultQueryRowLimit). finish = true; - return Response.ok().entity(targetDataSet).build(); + return FastLastHandler.fillLastValueDataSet(resultMap, defaultQueryRowLimit); } catch (Exception e) { finish = true; diff --git a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java index 02bdaf1c7fec9..996377dbead16 100644 --- a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/handler/QueryRowLimitUtilsTest.java @@ -32,13 +32,18 @@ public void resolveActualRowSizeLimitShouldUseConfiguredLimitAsHardLimit() { assertEquals(10, QueryRowLimitUtils.resolveActualRowSizeLimit(null, 10)); assertEquals(10, QueryRowLimitUtils.resolveActualRowSizeLimit(100, 10)); assertEquals(5, QueryRowLimitUtils.resolveActualRowSizeLimit(5, 10)); + // A caller-provided MAX_VALUE must be clamped down to the configured hard limit. + assertEquals(10, QueryRowLimitUtils.resolveActualRowSizeLimit(Integer.MAX_VALUE, 10)); } @Test - public void resolveActualRowSizeLimitShouldKeepHardLimitPositive() { - assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(null, 0)); - assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(100, 0)); - assertEquals(1, QueryRowLimitUtils.resolveActualRowSizeLimit(null, -1)); + public void resolveActualRowSizeLimitShouldFallBackToDefaultForNonPositiveConfig() { + // A non-positive rest_query_default_row_size_limit used to mean "unlimited"; it now falls back + // to the built-in default (10000) instead of being clamped down to a single row. + assertEquals(10000, QueryRowLimitUtils.resolveActualRowSizeLimit(null, 0)); + // A user request below the cap is still honored: min(100, default 10000) == 100. + assertEquals(100, QueryRowLimitUtils.resolveActualRowSizeLimit(100, 0)); + assertEquals(10000, QueryRowLimitUtils.resolveActualRowSizeLimit(null, -1)); } @Test @@ -47,6 +52,8 @@ public void exceedsLimitShouldRejectOnlyRowsBeyondTheHardLimit() { assertTrue(QueryRowLimitUtils.exceedsLimit(2, 1, 2)); assertTrue(QueryRowLimitUtils.exceedsLimit(0, 2, 1)); assertFalse(QueryRowLimitUtils.exceedsLimit(0, 0, 1)); - assertTrue(QueryRowLimitUtils.exceedsLimit(0, 2, 0)); + // A non-positive limit falls back to the default (10000), so small batches do not exceed it. + assertFalse(QueryRowLimitUtils.exceedsLimit(0, 2, 0)); + assertTrue(QueryRowLimitUtils.exceedsLimit(0, 10001, 0)); } } diff --git a/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandlerTest.java b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandlerTest.java new file mode 100644 index 0000000000000..a275813c39381 --- /dev/null +++ b/external-service-impl/rest/src/test/java/org/apache/iotdb/rest/protocol/v2/handler/FastLastHandlerTest.java @@ -0,0 +1,119 @@ +/* + * 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.rest.protocol.v2.handler; + +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache; +import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableId; +import org.apache.iotdb.rest.protocol.model.ExecutionStatus; +import org.apache.iotdb.rest.protocol.v2.model.QueryDataSet; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.read.TimeValuePair; +import org.apache.tsfile.utils.Pair; +import org.apache.tsfile.utils.TsPrimitiveType; +import org.junit.Test; + +import jakarta.ws.rs.core.Response; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class FastLastHandlerTest { + + private static final String DEVICE = "root.sg25.d1"; + + @Test + public void fillLastValueDataSetShouldReturnAllEntriesWhenUnderLimit() { + Map>>> resultMap = + newResultMap("s1", "s2", "s3"); + + Response response = FastLastHandler.fillLastValueDataSet(resultMap, 5); + + assertTrue(response.getEntity() instanceof QueryDataSet); + QueryDataSet dataSet = (QueryDataSet) response.getEntity(); + assertEquals(3, dataSet.getTimestamps().size()); + // [timeseries, valueList, dataTypeList], each holding one row per measurement + assertEquals(3, dataSet.getValues().size()); + assertEquals(3, dataSet.getValues().get(0).size()); + } + + @Test + public void fillLastValueDataSetShouldCapAtLimitAndReportError() { + // Regression for the fastLastQuery cache-hit path: a warm cache with more entries than the + // configured limit must NOT materialize the whole result into heap. + Map>>> resultMap = + newResultMap("s1", "s2", "s3", "s4", "s5"); + + Response response = FastLastHandler.fillLastValueDataSet(resultMap, 2); + + assertTrue(response.getEntity() instanceof ExecutionStatus); + ExecutionStatus status = (ExecutionStatus) response.getEntity(); + assertEquals(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode(), status.getCode().intValue()); + } + + @Test + public void fillLastValueDataSetShouldSkipPlaceholderAndNullEntries() { + Map>>> resultMap = + newResultMap("s1", "placeholder", "nullValue", "s4"); + + Response response = FastLastHandler.fillLastValueDataSet(resultMap, 10); + + assertTrue(response.getEntity() instanceof QueryDataSet); + QueryDataSet dataSet = (QueryDataSet) response.getEntity(); + // Only the two real entries (s1, s4) are materialized; placeholder/null entries are skipped. + assertEquals(2, dataSet.getTimestamps().size()); + } + + /** Builds a single-table/single-device resultMap; the special names insert sentinel entries. */ + private static Map>>> + newResultMap(String... measurementNames) { + Map> measurements = new HashMap<>(); + long timestamp = 100L; + for (String name : measurementNames) { + measurements.put(name, newEntry(name, timestamp++)); + } + Map>>> resultMap = + new HashMap<>(); + resultMap.put( + new TableId("root", "sg25"), + new HashMap<>( + Collections.singletonMap( + IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE), measurements))); + return resultMap; + } + + private static Pair newEntry(String name, long timestamp) { + if ("placeholder".equals(name)) { + return new Pair<>(TSDataType.INT64, TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN); + } + if ("nullValue".equals(name)) { + return new Pair<>(TSDataType.INT64, new TimeValuePair(timestamp, null)); + } + return new Pair<>( + TSDataType.INT64, + new TimeValuePair(timestamp, TsPrimitiveType.getByType(TSDataType.INT64, timestamp))); + } +} 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 91e84e89a0ece..2aa5b1a412eb1 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 @@ -609,6 +609,7 @@ enable_swagger=false # The maximum row limit for REST and Grafana query responses. # The request rowLimit/row_limit value cannot exceed this limit. +# A non-positive value is invalid and falls back to the default (10000); it no longer means unlimited. # effectiveMode: restart # Datatype: int rest_query_default_row_size_limit=10000