This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 9dd1bb8dbe3 Improve metadata OOM diagnostics (#18039)
9dd1bb8dbe3 is described below
commit 9dd1bb8dbe35b661de0a979693f52c61515d838d
Author: Caideyipi <[email protected]>
AuthorDate: Mon Jul 6 10:16:36 2026 +0800
Improve metadata OOM diagnostics (#18039)
* Improve metadata OOM diagnostics
* Internationalize metadata OOM diagnostics
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 39 +++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 28 +++
.../db/queryengine/common/MPPQueryContext.java | 278 ++++++++++++++++++++-
.../common/schematree/ClusterSchemaTree.java | 8 +
.../queryengine/plan/analyze/AnalyzeVisitor.java | 8 +
.../queryengine/plan/analyze/ExpressionUtils.java | 2 +
.../analyze/schema/ClusterSchemaFetchExecutor.java | 9 +
.../db/queryengine/common/MPPQueryContextTest.java | 125 +++++++++
8 files changed, 494 insertions(+), 3 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 061b0497011..d9f5b182122 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -129,6 +129,45 @@ public final class DataNodeQueryMessages {
"Free more memory than has been reserved.";
public static final String ESTIMATED_MODS_TREE_SIZE_DECREASED =
"Estimated mods tree size decreased from %d to %d for TsFile %s.";
+ public static final String RESULT_SET_COLUMN_METADATA_MEMORY_NOT_ENOUGH =
+ "Not enough memory while analyzing metadata for query result columns. "
+ + "The result set has too many columns. "
+ + "Before the failure, IoTDB had matched %,d source columns for
result-column "
+ + "expansion, expanded %,d source columns, and generated %,d
result-set columns. "
+ + "%s"
+ + "Current series pagination is %s. "
+ + "Use SLIMIT/SOFFSET to reduce returned series%s, narrow the path
pattern, "
+ + "or increase query memory%s. "
+ + "Memory details: source-column memory for result expansion %s, "
+ + "generated-result-column memory %s, requested this time %s,
current free memory %s. "
+ + "Original error: %s";
+ public static final String RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY =
+ "The matched source columns exceed the estimated current memory capacity
by "
+ + "at least %,d columns. ";
+ public static final String SCHEMA_FETCH_METADATA_MEMORY_NOT_ENOUGH =
+ "Not enough memory while fetching metadata for query analysis. "
+ + "The result set may have too many columns. "
+ + "Before the failure, IoTDB had deserialized %,d time-series
columns from schema "
+ + "fetch results. Schema fetch memory may be reserved before safely
deserializing "
+ + "the whole fetched metadata, so this count can be lower than the
matched schema "
+ + "columns. %s"
+ + "Current series pagination is %s. "
+ + "Use SLIMIT/SOFFSET to reduce returned series%s, narrow the path
pattern, "
+ + "or increase query memory%s. "
+ + "Memory details: fetched schema tree estimated memory %s, "
+ + "fetched schema tree reserved memory %s, requested this time %s, "
+ + "current free memory %s. Original error: %s";
+ public static final String SCHEMA_FETCH_COLUMNS_EXCEED_MEMORY_CAPACITY =
+ "The fetched schema columns exceed the estimated current memory capacity
by "
+ + "at least %,d columns. ";
+ public static final String USE_ALIGN_BY_DEVICE_TO_REDUCE_RESULT_COLUMNS =
+ ", use ALIGN BY DEVICE to reduce cross-device result columns";
+ public static final String BY_AT_LEAST_MEMORY_SIZE = " by at least %s";
+ public static final String FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL =
+ " for the query engine/operator memory pool";
+ public static final String SERIES_PAGINATION_FOR_DIAGNOSTICS = "SLIMIT=%s,
SOFFSET=%,d";
+ public static final String NOT_SET = "not set";
+ public static final String UNKNOWN = "unknown";
// --- Execution / Operator ---
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 1d7df89e027..4ca8eda2fb0 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -129,6 +129,34 @@ public final class DataNodeQueryMessages {
"释放的内存超过已预留的量。";
public static final String ESTIMATED_MODS_TREE_SIZE_DECREASED =
"估算的 mods tree 大小从 %d 减少到 %d，TsFile：%s。";
+ public static final String RESULT_SET_COLUMN_METADATA_MEMORY_NOT_ENOUGH =
+ "查询结果列元数据分析时内存不足。结果集包含过多列。"
+ + "失败前，IoTDB 已匹配 %,d 个用于结果列展开的源列，展开 %,d 个源列，并生成 %,d 个结果集列。"
+ + "%s"
+ + "当前序列分页为 %s。"
+ + "请使用 SLIMIT/SOFFSET 减少返回序列%s，缩小路径模式，或增加查询内存%s。"
+ + "内存详情：结果展开的源列内存 %s，生成结果列内存 %s，本次请求内存 %s，当前空闲内存 %s。"
+ + "原始错误：%s";
+ public static final String RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY =
+ "匹配的源列超过了当前估算内存容量，至少超出 %,d 列。";
+ public static final String SCHEMA_FETCH_METADATA_MEMORY_NOT_ENOUGH =
+ "查询分析拉取元数据时内存不足。结果集可能包含过多列。"
+ + "失败前，IoTDB 已从 schema 拉取结果中反序列化 %,d 个时间序列列。"
+ + "Schema 拉取内存可能会在安全反序列化完整元数据前预留，因此该计数可能小于匹配的 schema 列数。%s"
+ + "当前序列分页为 %s。"
+ + "请使用 SLIMIT/SOFFSET 减少返回序列%s，缩小路径模式，或增加查询内存%s。"
+ + "内存详情：拉取的 schema tree 估算内存 %s，拉取的 schema tree 预留内存 %s，"
+ + "本次请求内存 %s，当前空闲内存 %s。原始错误：%s";
+ public static final String SCHEMA_FETCH_COLUMNS_EXCEED_MEMORY_CAPACITY =
+ "拉取的 schema 列超过了当前估算内存容量，至少超出 %,d 列。";
+ public static final String USE_ALIGN_BY_DEVICE_TO_REDUCE_RESULT_COLUMNS =
+ "，使用 ALIGN BY DEVICE 可减少跨设备结果列";
+ public static final String BY_AT_LEAST_MEMORY_SIZE = "至少 %s";
+ public static final String FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL =
+ "（用于查询引擎/operator 内存池）";
+ public static final String SERIES_PAGINATION_FOR_DIAGNOSTICS =
"SLIMIT=%s，SOFFSET=%,d";
+ public static final String NOT_SET = "未设置";
+ public static final String UNKNOWN = "未知";
// --- Execution / Operator ---
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
index f0018845615..6412d334903 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/MPPQueryContext.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.queryengine.common;
+import org.apache.iotdb.calc.exception.MemoryNotEnoughException;
import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
@@ -44,6 +45,7 @@ import
org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils;
import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.analyze.lock.SchemaLockType;
+import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.NotThreadSafeMemoryReservationManager;
import
org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource;
import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat;
@@ -62,6 +64,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -143,6 +146,21 @@ public class MPPQueryContext implements IAuditEntity {
private boolean releaseSchemaTreeAfterAnalyzing = true;
private LongConsumer reserveMemoryForSchemaTreeFunc = null;
+ private boolean reservingMemoryForSchemaTree = false;
+
+ private boolean resultSetColumnMemoryTrackingEnabled = false;
+ private boolean alignByDeviceForResultSetColumnTracking = false;
+ private long seriesLimitForResultSetColumnTracking = 0;
+ private long seriesOffsetForResultSetColumnTracking = 0;
+ private long matchedSourceColumnsForResultSet = 0;
+ private long expandedSourceColumnsForResultSet = 0;
+ private long sourceColumnMemoryCostForResultSet = 0;
+ private long generatedResultSetColumns = 0;
+ private long generatedResultSetColumnMemoryCost = 0;
+ private long schemaFetchEstimatedMemoryCost = 0;
+ private long schemaFetchReservedMemoryCost = 0;
+ private long schemaFetchDeserializedColumnCount = 0;
+
private boolean userQuery = false;
/**
@@ -219,8 +237,17 @@ public class MPPQueryContext implements IAuditEntity {
if (reserveMemoryForSchemaTreeFunc == null) {
return;
}
- reserveMemoryForSchemaTreeFunc.accept(memoryCost);
+ schemaFetchEstimatedMemoryCost += memoryCost;
+ reservingMemoryForSchemaTree = true;
+ try {
+ reserveMemoryForSchemaTreeFunc.accept(memoryCost);
+ } catch (MemoryNotEnoughException e) {
+ throw enrichSchemaFetchMemoryNotEnoughException(e, memoryCost);
+ } finally {
+ reservingMemoryForSchemaTree = false;
+ }
this.reservedMemoryCostForSchemaTree += memoryCost;
+ this.schemaFetchReservedMemoryCost += memoryCost;
}
public void setReleaseSchemaTreeAfterAnalyzing(boolean
releaseSchemaTreeAfterAnalyzing) {
@@ -245,6 +272,7 @@ public class MPPQueryContext implements IAuditEntity {
}
this.initResultNodeContext();
this.releaseAllMemoryReservedForFrontEnd();
+ this.resetResultSetColumnMemoryTracking();
}
private void cleanUpCte() {
@@ -543,11 +571,25 @@ public class MPPQueryContext implements IAuditEntity {
* single-threaded manner.
*/
public void reserveMemoryForFrontEnd(final long bytes) {
- this.memoryReservationManager.reserveMemoryCumulatively(bytes);
+ try {
+ this.memoryReservationManager.reserveMemoryCumulatively(bytes);
+ } catch (MemoryNotEnoughException e) {
+ if (reservingMemoryForSchemaTree) {
+ throw e;
+ }
+ throw enrichResultSetColumnMemoryNotEnoughException(e, bytes);
+ }
}
public void reserveMemoryForFrontEndImmediately() {
- this.memoryReservationManager.reserveMemoryImmediately();
+ try {
+ this.memoryReservationManager.reserveMemoryImmediately();
+ } catch (MemoryNotEnoughException e) {
+ if (reservingMemoryForSchemaTree) {
+ throw e;
+ }
+ throw enrichResultSetColumnMemoryNotEnoughException(e,
extractRequestedMemory(e));
+ }
}
public void releaseAllMemoryReservedForFrontEnd() {
@@ -558,6 +600,236 @@ public class MPPQueryContext implements IAuditEntity {
this.memoryReservationManager.releaseMemoryCumulatively(bytes);
}
+ public void initResultSetColumnMemoryTracking(
+ long seriesLimit, long seriesOffset, boolean alignByDevice) {
+ resetResultSetColumnMemoryTracking();
+ resultSetColumnMemoryTrackingEnabled = true;
+ seriesLimitForResultSetColumnTracking = seriesLimit;
+ seriesOffsetForResultSetColumnTracking = seriesOffset;
+ alignByDeviceForResultSetColumnTracking = alignByDevice;
+ }
+
+ public void recordMatchedSourceColumnsForResultSet(long columnCount) {
+ if (resultSetColumnMemoryTrackingEnabled && columnCount > 0) {
+ matchedSourceColumnsForResultSet += columnCount;
+ }
+ }
+
+ public void recordExpandedSourceColumnForResultSet(long memoryCost) {
+ if (!resultSetColumnMemoryTrackingEnabled) {
+ return;
+ }
+ expandedSourceColumnsForResultSet++;
+ sourceColumnMemoryCostForResultSet += Math.max(memoryCost, 0);
+ }
+
+ public void recordGeneratedResultSetColumn(long memoryCost) {
+ if (!resultSetColumnMemoryTrackingEnabled) {
+ return;
+ }
+ generatedResultSetColumns++;
+ generatedResultSetColumnMemoryCost += Math.max(memoryCost, 0);
+ }
+
+ public void recordSchemaFetchDeserializedColumns(long columnCount) {
+ if (columnCount > 0) {
+ schemaFetchDeserializedColumnCount += columnCount;
+ }
+ }
+
+ private void resetResultSetColumnMemoryTracking() {
+ resultSetColumnMemoryTrackingEnabled = false;
+ alignByDeviceForResultSetColumnTracking = false;
+ seriesLimitForResultSetColumnTracking = 0;
+ seriesOffsetForResultSetColumnTracking = 0;
+ matchedSourceColumnsForResultSet = 0;
+ expandedSourceColumnsForResultSet = 0;
+ sourceColumnMemoryCostForResultSet = 0;
+ generatedResultSetColumns = 0;
+ generatedResultSetColumnMemoryCost = 0;
+ schemaFetchEstimatedMemoryCost = 0;
+ schemaFetchReservedMemoryCost = 0;
+ schemaFetchDeserializedColumnCount = 0;
+ }
+
+ private MemoryNotEnoughException
enrichResultSetColumnMemoryNotEnoughException(
+ MemoryNotEnoughException e, long requestedBytes) {
+ if (!resultSetColumnMemoryTrackingEnabled
+ || (matchedSourceColumnsForResultSet == 0
+ && expandedSourceColumnsForResultSet == 0
+ && generatedResultSetColumns == 0)) {
+ return e;
+ }
+
+ long freeBytes =
LocalExecutionPlanner.getInstance().getFreeMemoryForOperators();
+ long shortageBytes =
+ requestedBytes > 0 && requestedBytes > freeBytes ? requestedBytes -
freeBytes : -1;
+ long exceededColumns = estimateExceededColumns(freeBytes, requestedBytes);
+
+ return new MemoryNotEnoughException(
+ String.format(
+ Locale.ROOT,
+ DataNodeQueryMessages.RESULT_SET_COLUMN_METADATA_MEMORY_NOT_ENOUGH,
+ matchedSourceColumnsForResultSet,
+ expandedSourceColumnsForResultSet,
+ generatedResultSetColumns,
+ exceededColumns > 0
+ ? String.format(
+ Locale.ROOT,
+
DataNodeQueryMessages.RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY,
+ exceededColumns)
+ : "",
+ formatSeriesPaginationForDiagnostics(),
+ alignByDeviceForResultSetColumnTracking
+ ? ""
+ :
DataNodeQueryMessages.USE_ALIGN_BY_DEVICE_TO_REDUCE_RESULT_COLUMNS,
+ shortageBytes > 0
+ ? String.format(
+ Locale.ROOT,
+ DataNodeQueryMessages.BY_AT_LEAST_MEMORY_SIZE,
+ formatBytes(shortageBytes))
+ : DataNodeQueryMessages.FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL,
+ formatBytes(sourceColumnMemoryCostForResultSet),
+ formatBytes(generatedResultSetColumnMemoryCost),
+ formatBytes(requestedBytes),
+ formatBytes(freeBytes),
+ e.getMessage()));
+ }
+
+ private MemoryNotEnoughException enrichSchemaFetchMemoryNotEnoughException(
+ MemoryNotEnoughException e, long requestedBytes) {
+ long freeBytes =
LocalExecutionPlanner.getInstance().getFreeMemoryForOperators();
+ if (!resultSetColumnMemoryTrackingEnabled &&
schemaFetchDeserializedColumnCount == 0) {
+ return e;
+ }
+
+ long shortageBytes =
+ requestedBytes > 0 && requestedBytes > freeBytes ? requestedBytes -
freeBytes : -1;
+ long exceededColumns = estimateExceededSchemaFetchColumns(freeBytes,
requestedBytes);
+
+ return new MemoryNotEnoughException(
+ String.format(
+ Locale.ROOT,
+ DataNodeQueryMessages.SCHEMA_FETCH_METADATA_MEMORY_NOT_ENOUGH,
+ schemaFetchDeserializedColumnCount,
+ exceededColumns > 0
+ ? String.format(
+ Locale.ROOT,
+
DataNodeQueryMessages.SCHEMA_FETCH_COLUMNS_EXCEED_MEMORY_CAPACITY,
+ exceededColumns)
+ : "",
+ formatSeriesPaginationForDiagnostics(),
+ alignByDeviceForResultSetColumnTracking
+ ? ""
+ :
DataNodeQueryMessages.USE_ALIGN_BY_DEVICE_TO_REDUCE_RESULT_COLUMNS,
+ shortageBytes > 0
+ ? String.format(
+ Locale.ROOT,
+ DataNodeQueryMessages.BY_AT_LEAST_MEMORY_SIZE,
+ formatBytes(shortageBytes))
+ : DataNodeQueryMessages.FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL,
+ formatBytes(schemaFetchEstimatedMemoryCost),
+ formatBytes(schemaFetchReservedMemoryCost),
+ formatBytes(requestedBytes),
+ formatBytes(freeBytes),
+ e.getMessage()));
+ }
+
+ private long estimateExceededColumns(long freeBytes, long requestedBytes) {
+ long avgColumnMemory;
+ if (expandedSourceColumnsForResultSet > 0 &&
sourceColumnMemoryCostForResultSet > 0) {
+ avgColumnMemory =
+ Math.max(1, sourceColumnMemoryCostForResultSet /
expandedSourceColumnsForResultSet);
+ } else if (requestedBytes > 0) {
+ avgColumnMemory = requestedBytes;
+ } else {
+ return -1;
+ }
+ long estimatedCapacity =
+ (sourceColumnMemoryCostForResultSet + Math.max(freeBytes, 0)) /
avgColumnMemory;
+ long columnsToCompare =
+ Math.max(matchedSourceColumnsForResultSet,
expandedSourceColumnsForResultSet + 1);
+ return Math.max(0, columnsToCompare - estimatedCapacity);
+ }
+
+ private long estimateExceededSchemaFetchColumns(long freeBytes, long
requestedBytes) {
+ if (schemaFetchDeserializedColumnCount <= 0) {
+ return -1;
+ }
+
+ long avgColumnMemory;
+ long columnsToCompare = schemaFetchDeserializedColumnCount;
+ if (schemaFetchReservedMemoryCost > 0) {
+ avgColumnMemory =
+ Math.max(
+ 1, divideCeil(schemaFetchReservedMemoryCost,
schemaFetchDeserializedColumnCount));
+ if (requestedBytes > 0) {
+ columnsToCompare += Math.max(1, divideCeil(requestedBytes,
avgColumnMemory));
+ }
+ } else if (requestedBytes > 0) {
+ avgColumnMemory = Math.max(1, divideCeil(requestedBytes,
schemaFetchDeserializedColumnCount));
+ } else {
+ return -1;
+ }
+
+ long estimatedCapacity =
+ (schemaFetchReservedMemoryCost + Math.max(freeBytes, 0)) /
avgColumnMemory;
+ return Math.max(0, columnsToCompare - estimatedCapacity);
+ }
+
+ private static long divideCeil(long dividend, long divisor) {
+ return dividend / divisor + (dividend % divisor == 0 ? 0 : 1);
+ }
+
+ private String formatSeriesPaginationForDiagnostics() {
+ return String.format(
+ Locale.ROOT,
+ DataNodeQueryMessages.SERIES_PAGINATION_FOR_DIAGNOSTICS,
+ seriesLimitForResultSetColumnTracking > 0
+ ? String.format(Locale.ROOT, "%,d",
seriesLimitForResultSetColumnTracking)
+ : DataNodeQueryMessages.NOT_SET,
+ seriesOffsetForResultSetColumnTracking);
+ }
+
+ private static long extractRequestedMemory(MemoryNotEnoughException e) {
+ String message = e.getMessage();
+ if (message == null) {
+ return -1;
+ }
+ String marker = "the memory requested this time is ";
+ int start = message.indexOf(marker);
+ if (start < 0) {
+ return -1;
+ }
+ start += marker.length();
+ int end = message.indexOf('B', start);
+ if (end < 0) {
+ return -1;
+ }
+ try {
+ return Long.parseLong(message.substring(start, end));
+ } catch (NumberFormatException ignored) {
+ return -1;
+ }
+ }
+
+ private static String formatBytes(long bytes) {
+ if (bytes < 0) {
+ return DataNodeQueryMessages.UNKNOWN;
+ }
+ if (bytes < 1024) {
+ return bytes + " B";
+ }
+ double value = bytes;
+ String[] units = {"B", "KB", "MB", "GB", "TB"};
+ int unitIndex = 0;
+ while (value >= 1024 && unitIndex < units.length - 1) {
+ value /= 1024;
+ unitIndex++;
+ }
+ return String.format(Locale.ROOT, "%.2f %s (%d B)", value,
units[unitIndex], bytes);
+ }
+
public boolean useSampledAvgTimeseriesOperandMemCost() {
return numsOfSampledTimeseriesOperand >=
MIN_SIZE_TO_USE_SAMPLED_TIMESERIES_OPERAND_MEM_COST;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/schematree/ClusterSchemaTree.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/schematree/ClusterSchemaTree.java
index 632c86cc2b6..26df77e6823 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/schematree/ClusterSchemaTree.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/common/schematree/ClusterSchemaTree.java
@@ -570,10 +570,16 @@ public class ClusterSchemaTree implements ISchemaTree {
private Map<Integer, Template> templateMap = new HashMap<>();
private boolean isFirstBatch = true;
+ private long measurementCount = 0;
+
public boolean isFirstBatch() {
return isFirstBatch;
}
+ public long getMeasurementCount() {
+ return measurementCount;
+ }
+
public void deserializeFromBatch(InputStream inputStream) throws
IOException {
isFirstBatch = false;
while (inputStream.available() > 0) {
@@ -581,6 +587,7 @@ public class ClusterSchemaTree implements ISchemaTree {
if (nodeType == SCHEMA_MEASUREMENT_NODE) {
SchemaMeasurementNode measurementNode =
SchemaMeasurementNode.deserialize(inputStream);
stack.push(measurementNode);
+ measurementCount++;
if (measurementNode.isLogicalView()) {
hasLogicalView = true;
}
@@ -638,6 +645,7 @@ public class ClusterSchemaTree implements ISchemaTree {
// templateMap is set to the returned schema tree, so we should create a
new one
templateMap = new HashMap<>();
isFirstBatch = true;
+ measurementCount = 0;
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
index 9e67239d50d..d8f1781bdaf 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeVisitor.java
@@ -282,6 +282,10 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
// check for semantic errors
queryStatement.semanticCheck();
+ context.initResultSetColumnMemoryTracking(
+ queryStatement.getSeriesLimit(),
+ queryStatement.getSeriesOffset(),
+ queryStatement.isAlignByDevice());
// fetch model inference information and check
analyzeModelInference(analysis, queryStatement);
@@ -721,6 +725,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
groupByLevelHelper.applyLevels(
isCountStar, resultExpression, resultColumn.getAlias(),
analysis);
Expression normalizedOutputExpression =
normalizeExpression(outputExpression);
+
queryContext.recordGeneratedResultSetColumn(normalizedOutputExpression.ramBytesUsed());
analyzeExpressionType(analysis, normalizedOutputExpression);
outputExpressionSet.add(
new Pair<>(
@@ -785,6 +790,7 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
checkAliasUniqueness(resultColumn.getAlias(), aliasSet);
Expression normalizedExpression =
normalizeExpression(resultExpression);
+
queryContext.recordGeneratedResultSetColumn(normalizedExpression.ramBytesUsed());
analyzeExpressionType(analysis, normalizedExpression);
outputExpressions.add(
new Pair<>(
@@ -884,6 +890,8 @@ public class AnalyzeVisitor extends
StatementVisitor<Analysis, MPPQueryContext>
Expression lowerCaseMeasurementExpression =
toLowerCaseExpression(measurementExpression);
analyzeExpressionType(analysis, lowerCaseMeasurementExpression);
+ queryContext.recordGeneratedResultSetColumn(
+ lowerCaseMeasurementExpression.ramBytesUsed());
outputExpressions.add(
new Pair<>(
lowerCaseMeasurementExpression,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
index 9c7e3ce7a57..9fe0bb15b19 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/ExpressionUtils.java
@@ -71,6 +71,7 @@ public class ExpressionUtils {
final List<? extends PartialPath> actualPaths,
final MPPQueryContext queryContext) {
List<Expression> resultExpressions = new ArrayList<>();
+ queryContext.recordMatchedSourceColumnsForResultSet(actualPaths.size());
for (PartialPath actualPath : actualPaths) {
Expression expression = reconstructTimeSeriesOperand(rawExpression,
actualPath);
long memCost;
@@ -82,6 +83,7 @@ public class ExpressionUtils {
}
queryContext.reserveMemoryForFrontEnd(memCost);
+ queryContext.recordExpandedSourceColumnForResultSet(memCost);
resultExpressions.add(expression);
}
return resultExpressions;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
index a5f0856d74f..a092f4b54f9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/schema/ClusterSchemaFetchExecutor.java
@@ -61,6 +61,8 @@ import java.util.Optional;
import java.util.Set;
import java.util.function.Consumer;
+import static org.apache.iotdb.commons.schema.SchemaConstant.ALL_MATCH_PATTERN;
+
class ClusterSchemaFetchExecutor {
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -336,6 +338,8 @@ class ClusterSchemaFetchExecutor {
// for data from old version
ClusterSchemaTree deserializedSchemaTree =
ClusterSchemaTree.deserialize(inputStream);
if (context != null) {
+ context.recordSchemaFetchDeserializedColumns(
+
deserializedSchemaTree.searchMeasurementPaths(ALL_MATCH_PATTERN).left.size());
context.reserveMemoryForSchemaTree(deserializedSchemaTree.ramBytesUsed());
}
resultSchemaTree.mergeSchemaTree(deserializedSchemaTree);
@@ -346,7 +350,12 @@ class ClusterSchemaFetchExecutor {
context.reserveMemoryForSchemaTree(memCost);
}
}
+ long measurementCountBeforeDeserialization =
deserializer.getMeasurementCount();
deserializer.deserializeFromBatch(inputStream);
+ if (context != null) {
+ context.recordSchemaFetchDeserializedColumns(
+ deserializer.getMeasurementCount() -
measurementCountBeforeDeserialization);
+ }
if (type == 3) {
// 'type == 3' indicates this batch is finished
resultSchemaTree.mergeSchemaTree(deserializer.finish());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/common/MPPQueryContextTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/common/MPPQueryContextTest.java
new file mode 100644
index 00000000000..9f5803e1930
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/common/MPPQueryContextTest.java
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.common;
+
+import org.apache.iotdb.calc.exception.MemoryNotEnoughException;
+import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class MPPQueryContextTest {
+
+ private static final long MEMORY_BATCH_THRESHOLD = 1024L * 1024L;
+
+ @Test
+ public void
resultSetColumnMemoryNotEnoughExceptionContainsColumnDiagnostics() {
+ MPPQueryContext context = new MPPQueryContext(new
QueryId("result_column_oom_test"));
+ long failedRequestBytes = requestLargerThanFreeOperatorMemory();
+
+ context.initResultSetColumnMemoryTracking(100, 5, false);
+ context.recordMatchedSourceColumnsForResultSet(10);
+ context.recordExpandedSourceColumnForResultSet(failedRequestBytes);
+ context.recordGeneratedResultSetColumn(128);
+
+ MemoryNotEnoughException exception =
+ Assert.assertThrows(
+ MemoryNotEnoughException.class,
+ () -> context.reserveMemoryForFrontEnd(failedRequestBytes));
+
+ String message = exception.getMessage();
+ assertContains(message, "Not enough memory while analyzing metadata for
query result columns.");
+ assertContains(message, "The result set has too many columns.");
+ assertContains(message, "matched 10 source columns");
+ assertContains(message, "expanded 1 source columns");
+ assertContains(message, "generated 1 result-set columns");
+ assertContains(message, "exceed the estimated current memory capacity by
at least");
+ assertContains(message, "SLIMIT=100, SOFFSET=5");
+ assertContains(message, "Use SLIMIT/SOFFSET");
+ assertContains(message, "ALIGN BY DEVICE");
+ assertContains(message, "increase query memory by at least");
+ assertContains(message, "requested this time");
+ assertContains(message, "current free memory");
+ assertContains(message, "Original error:");
+ }
+
+ @Test
+ public void
schemaFetchMemoryNotEnoughExceptionContainsFetchedColumnDiagnostics() {
+ MPPQueryContext context = new MPPQueryContext(new
QueryId("schema_fetch_oom_test"));
+ long failedRequestBytes = requestLargerThanFreeOperatorMemory();
+
+ context.initResultSetColumnMemoryTracking(0, 2, true);
+ context.recordSchemaFetchDeserializedColumns(4);
+ context.setReserveMemoryForSchemaTreeFunc(
+ bytes -> {
+ throw new MemoryNotEnoughException("schema fetch OOM");
+ });
+
+ MemoryNotEnoughException exception =
+ Assert.assertThrows(
+ MemoryNotEnoughException.class,
+ () -> context.reserveMemoryForSchemaTree(failedRequestBytes));
+
+ String message = exception.getMessage();
+ assertContains(message, "Not enough memory while fetching metadata for
query analysis.");
+ assertContains(message, "deserialized 4 time-series columns");
+ assertContains(message, "fetched schema columns exceed the estimated
current memory capacity");
+ assertContains(message, "SLIMIT=not set, SOFFSET=2");
+ assertContains(message, "Use SLIMIT/SOFFSET");
+ assertContains(message, "increase query memory by at least");
+ assertContains(message, "fetched schema tree estimated memory");
+ assertContains(message, "requested this time");
+ assertContains(message, "schema fetch OOM");
+ }
+
+ @Test
+ public void
schemaFetchMemoryNotEnoughExceptionKeepsOriginalWithoutColumnContext() {
+ MPPQueryContext context = new MPPQueryContext(new
QueryId("schema_fetch_without_context"));
+ MemoryNotEnoughException expected = new MemoryNotEnoughException("original
schema OOM");
+ context.setReserveMemoryForSchemaTreeFunc(
+ bytes -> {
+ throw expected;
+ });
+
+ MemoryNotEnoughException actual =
+ Assert.assertThrows(
+ MemoryNotEnoughException.class, () ->
context.reserveMemoryForSchemaTree(1));
+
+ Assert.assertSame(expected, actual);
+ Assert.assertEquals("original schema OOM", actual.getMessage());
+ }
+
+ private static long requestLargerThanFreeOperatorMemory() {
+ long freeBytes =
LocalExecutionPlanner.getInstance().getFreeMemoryForOperators();
+ if (freeBytes < MEMORY_BATCH_THRESHOLD) {
+ return MEMORY_BATCH_THRESHOLD;
+ }
+ if (freeBytes >= Long.MAX_VALUE - 1) {
+ return Long.MAX_VALUE;
+ }
+ return freeBytes + 1;
+ }
+
+ private static void assertContains(String message, String expected) {
+ Assert.assertTrue(
+ String.format("Expected message to contain <%s>, but was <%s>.",
expected, message),
+ message.contains(expected));
+ }
+}
