This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new f0e8ba2637f [To dev/1.3] Enforce REST query row limit as hard cap
(#18149) (#18197)
f0e8ba2637f is described below
commit f0e8ba2637f2ae50b7ff3e7697240c9ff68a81df
Author: Jackie Tien <[email protected]>
AuthorDate: Tue Jul 14 12:17:22 2026 +0800
[To dev/1.3] Enforce REST query row limit as hard cap (#18149) (#18197)
---
.../src/test/resources/iotdb-common.properties | 3 +-
.../protocol/rest/handler/QueryRowLimitUtils.java | 68 +++++++
.../rest/v1/handler/QueryDataSetHandler.java | 114 ++++++------
.../rest/v1/impl/GrafanaApiServiceImpl.java | 19 +-
.../protocol/rest/v1/impl/RestApiServiceImpl.java | 8 +-
.../protocol/rest/v2/handler/FastLastHandler.java | 59 +++++-
.../rest/v2/handler/QueryDataSetHandler.java | 114 ++++++------
.../rest/v2/impl/GrafanaApiServiceImpl.java | 19 +-
.../protocol/rest/v2/impl/RestApiServiceImpl.java | 43 +----
.../rest/handler/QueryRowLimitUtilsTest.java | 59 ++++++
.../rest/v1/handler/QueryDataSetHandlerTest.java | 200 +++++++++++++++++++++
.../rest/v2/handler/FastLastHandlerTest.java | 108 +++++++++++
.../src/test/resources/iotdb-common.properties | 3 +-
.../src/test/resources/iotdb-system.properties | 3 +-
.../conf/iotdb-system.properties.template | 4 +-
.../openapi/src/main/openapi3/iotdb_rest_v1.yaml | 1 +
.../openapi/src/main/openapi3/iotdb_rest_v2.yaml | 1 +
17 files changed, 652 insertions(+), 174 deletions(-)
diff --git a/iotdb-client/session/src/test/resources/iotdb-common.properties
b/iotdb-client/session/src/test/resources/iotdb-common.properties
index 965b3c010ed..ec9b6914ef8 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 swagger. eg:
http://ip:port/swagger.json
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtils.java
new file mode 100644
index 00000000000..00f8a993ae8
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtils.java
@@ -0,0 +1,68 @@
+/*
+ * 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.protocol.rest.handler;
+
+import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import javax.ws.rs.core.Response;
+
+public final class QueryRowLimitUtils {
+
+ /**
+ * 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() {}
+
+ 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 rowSizeLimit > 0 ? rowSizeLimit : DEFAULT_ROW_SIZE_LIMIT;
+ }
+
+ 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(
+ "Dataset row size exceeded the given max row size
(%d)", rowSizeLimit)))
+ .build();
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandler.java
index c82b07c0603..b87ace54df4 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandler.java
@@ -18,6 +18,7 @@
package org.apache.iotdb.db.protocol.rest.v1.handler;
import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.v1.model.ExecutionStatus;
import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
@@ -47,7 +48,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)
@@ -136,7 +137,6 @@ public class QueryDataSetHandler {
final long timePrecision)
throws IoTDBException {
int fetched = 0;
- int columnNum = queryExecution.getOutputValueColumnCount();
DatasetHeader header = queryExecution.getDatasetHeader();
List<String> resultColumns = header.getRespColumns();
@@ -147,17 +147,6 @@ public class QueryDataSetHandler {
}
while (true) {
- if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) {
- return Response.ok()
- .entity(
- new org.apache.iotdb.db.protocol.rest.model.ExecutionStatus()
- .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode())
- .message(
- String.format(
- "Dataset row size exceeded the given max row size
(%d)",
- actualRowSizeLimit)))
- .build();
- }
Optional<TsBlock> optionalTsBlock = queryExecution.getBatchResult();
if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) {
if (fetched == 0) {
@@ -169,6 +158,9 @@ public class QueryDataSetHandler {
}
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(
@@ -180,7 +172,6 @@ public class QueryDataSetHandler {
Column column = tsBlock.getColumn(headerMap.get(resultColumns.get(k)));
List<Object> targetDataSetColumn = targetDataSet.getValues().get(k);
for (int i = 0; i < currentCount; i++) {
- fetched++;
if (column.isNull(i)) {
targetDataSetColumn.add(null);
} else {
@@ -190,10 +181,8 @@ public class QueryDataSetHandler {
: column.getObject(i));
}
}
- if (k != columnNum - 1) {
- fetched -= currentCount;
- }
}
+ fetched += currentCount;
}
return Response.ok().entity(targetDataSet).build();
}
@@ -207,17 +196,6 @@ public class QueryDataSetHandler {
int fetched = 0;
int columnNum = queryExecution.getOutputValueColumnCount();
while (true) {
- if (0 < actualRowSizeLimit && actualRowSizeLimit <= fetched) {
- return Response.ok()
- .entity(
- new org.apache.iotdb.db.protocol.rest.model.ExecutionStatus()
- .code(TSStatusCode.QUERY_PROCESS_ERROR.getStatusCode())
- .message(
- String.format(
- "Dataset row size exceeded the given max row size
(%d)",
- actualRowSizeLimit)))
- .build();
- }
Optional<TsBlock> optionalTsBlock = queryExecution.getBatchResult();
if (!optionalTsBlock.isPresent()) {
if (fetched == 0) {
@@ -232,11 +210,13 @@ public class QueryDataSetHandler {
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<Object> targetDataSetColumn = targetDataSet.getValues().get(k);
for (int i = 0; i < currentCount; i++) {
- fetched++;
if (column.isNull(i)) {
targetDataSetColumn.add(null);
} else {
@@ -246,53 +226,67 @@ public class QueryDataSetHandler {
: column.getObject(i));
}
}
- if (k != columnNum - 1) {
- fetched -= currentCount;
- }
}
+ fetched += currentCount;
}
return Response.ok().entity(targetDataSet).build();
}
public static Response fillGrafanaVariablesResult(
- IQueryExecution queryExecution, Statement statement) throws
IoTDBException {
+ IQueryExecution queryExecution, Statement statement, int
actualRowSizeLimit)
+ throws IoTDBException {
List<String> results = new ArrayList<>();
- Optional<TsBlock> 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<TsBlock> 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<String> nodes = new ArrayList<>();
- Optional<TsBlock> 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<TsBlock> 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
index bd955fd00da..8ba6eb44337 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/GrafanaApiServiceImpl.java
@@ -21,7 +21,9 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
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.rest.handler.AuthorizationHandler;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.v1.GrafanaApiService;
import org.apache.iotdb.db.protocol.rest.v1.handler.ExceptionHandler;
import org.apache.iotdb.db.protocol.rest.v1.handler.QueryDataSetHandler;
@@ -67,11 +69,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":
@@ -130,7 +136,8 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
}
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();
@@ -200,9 +207,11 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
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) {
@@ -262,10 +271,10 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
index 084b7cc53e9..c50baa9bb54 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v1/impl/RestApiServiceImpl.java
@@ -21,6 +21,7 @@ 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.rest.handler.AuthorizationHandler;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.utils.InsertTabletSortDataUtils;
import org.apache.iotdb.db.protocol.rest.v1.RestApiService;
import org.apache.iotdb.db.protocol.rest.v1.handler.ExceptionHandler;
@@ -66,14 +67,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
@@ -208,7 +210,7 @@ public class RestApiServiceImpl extends RestApiService {
return QueryDataSetHandler.fillQueryDataSet(
queryExecution,
statement,
- sql.getRowLimit() == null ? defaultQueryRowLimit :
sql.getRowLimit());
+ QueryRowLimitUtils.resolveActualRowSizeLimit(sql.getRowLimit(),
defaultQueryRowLimit));
}
} catch (Exception e) {
finish = true;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandler.java
index 5d1c3d0109c..af1657a8b13 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandler.java
@@ -20,16 +20,25 @@
package org.apache.iotdb.db.protocol.rest.v2.handler;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.v2.model.ExecutionStatus;
import org.apache.iotdb.db.protocol.rest.v2.model.PrefixPathList;
+import org.apache.iotdb.db.protocol.rest.v2.model.QueryDataSet;
import org.apache.iotdb.db.protocol.session.IClientSession;
+import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
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.read.TimeValuePair;
+
import javax.ws.rs.core.Response;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.List;
+import java.util.Map;
public class FastLastHandler {
@@ -59,8 +68,7 @@ public class FastLastHandler {
.build();
}
- public static void setupTargetDataSet(
- org.apache.iotdb.db.protocol.rest.v2.model.QueryDataSet dataSet) {
+ public static void setupTargetDataSet(QueryDataSet dataSet) {
dataSet.addExpressionsItem("Timeseries");
dataSet.addExpressionsItem("Value");
dataSet.addExpressionsItem("DataType");
@@ -70,4 +78,51 @@ public class FastLastHandler {
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 device/measurement, as filled by
the schema cache
+ * @param actualRowSizeLimit hard cap on the number of returned rows
+ */
+ public static Response fillLastValueDataSet(
+ Map<PartialPath, Map<String, TimeValuePair>> resultMap, int
actualRowSizeLimit) {
+ QueryDataSet targetDataSet = new QueryDataSet();
+ setupTargetDataSet(targetDataSet);
+ List<Object> timeseries = new ArrayList<>();
+ List<Object> valueList = new ArrayList<>();
+ List<Object> dataTypeList = new ArrayList<>();
+ int fetched = 0;
+
+ for (final Map.Entry<PartialPath, Map<String, TimeValuePair>>
device2MeasurementLastEntry :
+ resultMap.entrySet()) {
+ final String deviceWithSeparator =
+ device2MeasurementLastEntry.getKey() + TsFileConstant.PATH_SEPARATOR;
+ for (Map.Entry<String, TimeValuePair> measurementEntry :
+ device2MeasurementLastEntry.getValue().entrySet()) {
+ final TimeValuePair tvPair = measurementEntry.getValue();
+ if (tvPair == null
+ || tvPair == DeviceLastCache.EMPTY_TIME_VALUE_PAIR
+ || 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 + measurementEntry.getKey());
+ fetched++;
+ }
+ }
+ if (!timeseries.isEmpty()) {
+ targetDataSet.addValuesItem(timeseries);
+ targetDataSet.addValuesItem(valueList);
+ targetDataSet.addValuesItem(dataTypeList);
+ }
+ return Response.ok().entity(targetDataSet).build();
+ }
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/QueryDataSetHandler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/QueryDataSetHandler.java
index f3f7d68dd0b..e77265abe1d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/QueryDataSetHandler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/handler/QueryDataSetHandler.java
@@ -18,6 +18,7 @@
package org.apache.iotdb.db.protocol.rest.v2.handler;
import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
@@ -47,7 +48,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)
@@ -138,7 +139,6 @@ public class QueryDataSetHandler {
final long timePrecision)
throws IoTDBException {
int fetched = 0;
- int columnNum = queryExecution.getOutputValueColumnCount();
DatasetHeader header = queryExecution.getDatasetHeader();
List<String> resultColumns = header.getRespColumns();
@@ -151,17 +151,6 @@ public class QueryDataSetHandler {
}
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<TsBlock> optionalTsBlock = queryExecution.getBatchResult();
if (!optionalTsBlock.isPresent() || optionalTsBlock.get().isEmpty()) {
if (fetched == 0) {
@@ -173,6 +162,9 @@ public class QueryDataSetHandler {
}
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(
@@ -184,7 +176,6 @@ public class QueryDataSetHandler {
Column column = tsBlock.getColumn(headerMap.get(resultColumns.get(k)));
List<Object> targetDataSetColumn = targetDataSet.getValues().get(k);
for (int i = 0; i < currentCount; i++) {
- fetched++;
if (column.isNull(i)) {
targetDataSetColumn.add(null);
} else {
@@ -194,10 +185,8 @@ public class QueryDataSetHandler {
: column.getObject(i));
}
}
- if (k != columnNum - 1) {
- fetched -= currentCount;
- }
}
+ fetched += currentCount;
}
return Response.ok().entity(targetDataSet).build();
}
@@ -211,17 +200,6 @@ public class QueryDataSetHandler {
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<TsBlock> optionalTsBlock = queryExecution.getBatchResult();
if (!optionalTsBlock.isPresent()) {
if (fetched == 0) {
@@ -236,11 +214,13 @@ public class QueryDataSetHandler {
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<Object> targetDataSetColumn = targetDataSet.getValues().get(k);
for (int i = 0; i < currentCount; i++) {
- fetched++;
if (column.isNull(i)) {
targetDataSetColumn.add(null);
} else {
@@ -250,53 +230,67 @@ public class QueryDataSetHandler {
: column.getObject(i));
}
}
- if (k != columnNum - 1) {
- fetched -= currentCount;
- }
}
+ fetched += currentCount;
}
return Response.ok().entity(targetDataSet).build();
}
public static Response fillGrafanaVariablesResult(
- IQueryExecution queryExecution, Statement statement) throws
IoTDBException {
+ IQueryExecution queryExecution, Statement statement, int
actualRowSizeLimit)
+ throws IoTDBException {
List<String> results = new ArrayList<>();
- Optional<TsBlock> 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<TsBlock> 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<String> nodes = new ArrayList<>();
- Optional<TsBlock> 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<TsBlock> 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/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
index 6a5ae13a0cb..b57306fce8c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/rest/v2/impl/GrafanaApiServiceImpl.java
@@ -21,7 +21,9 @@ import org.apache.iotdb.commons.conf.CommonDescriptor;
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.rest.handler.AuthorizationHandler;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.v2.GrafanaApiService;
import org.apache.iotdb.db.protocol.rest.v2.handler.ExceptionHandler;
import org.apache.iotdb.db.protocol.rest.v2.handler.QueryDataSetHandler;
@@ -67,11 +69,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":
@@ -130,7 +136,8 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
}
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();
@@ -200,9 +207,11 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
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) {
@@ -262,10 +271,10 @@ public class GrafanaApiServiceImpl extends
GrafanaApiService {
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/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 00309806342..80e86c1a998 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
@@ -23,6 +23,7 @@ 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.rest.handler.AuthorizationHandler;
+import org.apache.iotdb.db.protocol.rest.handler.QueryRowLimitUtils;
import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
import org.apache.iotdb.db.protocol.rest.utils.InsertTabletSortDataUtils;
import org.apache.iotdb.db.protocol.rest.v2.NotFoundException;
@@ -36,7 +37,6 @@ import
org.apache.iotdb.db.protocol.rest.v2.handler.StatementConstructionHandler
import org.apache.iotdb.db.protocol.rest.v2.model.InsertRecordsRequest;
import org.apache.iotdb.db.protocol.rest.v2.model.InsertTabletRequest;
import org.apache.iotdb.db.protocol.rest.v2.model.PrefixPathList;
-import org.apache.iotdb.db.protocol.rest.v2.model.QueryDataSet;
import org.apache.iotdb.db.protocol.rest.v2.model.SQL;
import org.apache.iotdb.db.protocol.session.IClientSession;
import org.apache.iotdb.db.protocol.session.SessionManager;
@@ -46,7 +46,6 @@ import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.analyze.ClusterPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeSchemaCache;
-import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
import
org.apache.iotdb.db.queryengine.plan.analyze.schema.ClusterSchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
@@ -63,14 +62,12 @@ import org.apache.iotdb.db.utils.SetThreadName;
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.read.TimeValuePair;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.SecurityContext;
import java.time.ZoneId;
-import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -89,14 +86,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
@@ -167,34 +165,9 @@ public class RestApiServiceImpl extends RestApiService {
}
}
- // Cache hit: build response directly
- QueryDataSet targetDataSet = new QueryDataSet();
-
- FastLastHandler.setupTargetDataSet(targetDataSet);
- List<Object> timeseries = new ArrayList<>();
- List<Object> valueList = new ArrayList<>();
- List<Object> dataTypeList = new ArrayList<>();
- for (final Map.Entry<PartialPath, Map<String, TimeValuePair>>
device2MeasurementLastEntry :
- resultMap.entrySet()) {
- final String deviceWithSeparator =
- device2MeasurementLastEntry.getKey() +
TsFileConstant.PATH_SEPARATOR;
- for (Map.Entry<String, TimeValuePair> measurementEntry :
- device2MeasurementLastEntry.getValue().entrySet()) {
- final TimeValuePair tvPair = measurementEntry.getValue();
- if (tvPair != DeviceLastCache.EMPTY_TIME_VALUE_PAIR) {
- valueList.add(tvPair.getValue().getStringValue());
- dataTypeList.add(tvPair.getValue().getDataType().name());
- targetDataSet.addTimestampsItem(tvPair.getTimestamp());
- timeseries.add(deviceWithSeparator + measurementEntry.getKey());
- }
- }
- }
- if (!timeseries.isEmpty()) {
- targetDataSet.addValuesItem(timeseries);
- targetDataSet.addValuesItem(valueList);
- targetDataSet.addValuesItem(dataTypeList);
- }
- return Response.ok().entity(targetDataSet).build();
+ // Cache hit: build response directly (capped by defaultQueryRowLimit).
+ finish = true;
+ return FastLastHandler.fillLastValueDataSet(resultMap,
defaultQueryRowLimit);
} catch (Exception e) {
finish = true;
@@ -340,7 +313,7 @@ public class RestApiServiceImpl extends RestApiService {
return QueryDataSetHandler.fillQueryDataSet(
queryExecution,
statement,
- sql.getRowLimit() == null ? defaultQueryRowLimit :
sql.getRowLimit());
+ QueryRowLimitUtils.resolveActualRowSizeLimit(sql.getRowLimit(),
defaultQueryRowLimit));
}
} catch (Exception e) {
finish = true;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtilsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtilsTest.java
new file mode 100644
index 00000000000..3f21e8ea167
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/handler/QueryRowLimitUtilsTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.protocol.rest.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));
+ // 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
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
+ 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));
+ // 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/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandlerTest.java
new file mode 100644
index 00000000000..9f5d3a097ea
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v1/handler/QueryDataSetHandlerTest.java
@@ -0,0 +1,200 @@
+/*
+ * 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.protocol.rest.v1.handler;
+
+import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
+import org.apache.iotdb.db.protocol.rest.v1.model.QueryDataSet;
+import org.apache.iotdb.db.queryengine.common.header.ColumnHeader;
+import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
+import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
+import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
+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 javax.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<Optional<TsBlock>> 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.setColumnToTsBlockIndexMap(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<TsBlock> getBatchResult() throws IoTDBException {
+ return batches.remove();
+ }
+
+ @Override
+ public Optional<ByteBuffer> 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 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<String> getExecuteSQL() {
+ return Optional.empty();
+ }
+
+ @Override
+ public String getStatementType() {
+ return null;
+ }
+
+ @Override
+ public String getClientHostname() {
+ return null;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandlerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandlerTest.java
new file mode 100644
index 00000000000..786db7492d1
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/protocol/rest/v2/handler/FastLastHandlerTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.protocol.rest.v2.handler;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.protocol.rest.model.ExecutionStatus;
+import org.apache.iotdb.db.protocol.rest.v2.model.QueryDataSet;
+import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DeviceLastCache;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.TimeValuePair;
+import org.apache.tsfile.utils.TsPrimitiveType;
+import org.junit.Test;
+
+import javax.ws.rs.core.Response;
+
+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()
throws Exception {
+ Map<PartialPath, Map<String, TimeValuePair>> 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() throws
Exception {
+ // 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<PartialPath, Map<String, TimeValuePair>> 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() throws
Exception {
+ Map<PartialPath, Map<String, TimeValuePair>> 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-device resultMap; the special names insert sentinel
entries. */
+ private static Map<PartialPath, Map<String, TimeValuePair>> newResultMap(
+ String... measurementNames) throws Exception {
+ Map<String, TimeValuePair> measurements = new HashMap<>();
+ long timestamp = 100L;
+ for (String name : measurementNames) {
+ measurements.put(name, newEntry(name, timestamp++));
+ }
+ Map<PartialPath, Map<String, TimeValuePair>> resultMap = new HashMap<>();
+ resultMap.put(new PartialPath(DEVICE), measurements);
+ return resultMap;
+ }
+
+ private static TimeValuePair newEntry(String name, long timestamp) {
+ if ("placeholder".equals(name)) {
+ return DeviceLastCache.EMPTY_TIME_VALUE_PAIR;
+ }
+ if ("nullValue".equals(name)) {
+ return new TimeValuePair(timestamp, null);
+ }
+ return new TimeValuePair(timestamp,
TsPrimitiveType.getByType(TSDataType.INT64, timestamp));
+ }
+}
diff --git a/iotdb-core/datanode/src/test/resources/iotdb-common.properties
b/iotdb-core/datanode/src/test/resources/iotdb-common.properties
index 1053eaafa2d..95ae09870f1 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 swagger. eg:
http://ip:port/swagger.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 4732caa9fee..6a19f218257 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 swagger. eg:
http://ip:port/swagger.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 42c0aac0e1a..bda26c3cb93 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
@@ -563,7 +563,9 @@ 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.
+# 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
diff --git a/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v1.yaml
b/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v1.yaml
index 18080c693ab..7149ddd1cd3 100644
--- a/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v1.yaml
+++ b/iotdb-protocol/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/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v2.yaml
b/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v2.yaml
index 0cae51bef23..0ebb430b1b9 100644
--- a/iotdb-protocol/openapi/src/main/openapi3/iotdb_rest_v2.yaml
+++ b/iotdb-protocol/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
