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 25bf38eba0c [Query] Fix result-column OOM diagnostics (#18661)
25bf38eba0c is described below
commit 25bf38eba0cc9d3d213b79068ea002e62c5ced96
Author: Caideyipi <[email protected]>
AuthorDate: Thu Sep 17 12:23:51 2026 +0800
[Query] Fix result-column OOM diagnostics (#18661)
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 3 ++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 2 +
.../db/queryengine/common/MPPQueryContext.java | 47 ++++++++++++++++--
.../plan/planner/LocalExecutionPlanner.java | 10 ++--
.../memory/OperatorMemoryNotEnoughException.java | 43 ++++++++++++++++
.../db/queryengine/common/MPPQueryContextTest.java | 58 ++++++++++++++++++++++
6 files changed, 155 insertions(+), 8 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 bf7a3601003..7b60fe71c2b 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
@@ -156,6 +156,9 @@ public final class DataNodeQueryMessages {
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 RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT =
+ "The failed memory reservation exceeds available memory by the
equivalent of at least "
+ + "%,d columns, estimated from the observed average column size. ";
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. "
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 f26f2306be2..eeefd24141d 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
@@ -152,6 +152,8 @@ public final class DataNodeQueryMessages {
+ "原始错误：%s";
public static final String RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY =
"匹配的源列超过了当前估算内存容量，至少超出 %,d 列。";
+ public static final String RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT =
+ "本次失败的内存申请超出可用内存，按已记录列的平均内存估算，至少超出相当于 %,d 列的容量。";
public static final String SCHEMA_FETCH_METADATA_MEMORY_NOT_ENOUGH =
"查询分析拉取元数据时内存不足。结果集可能包含过多列。"
+ "失败前，IoTDB 已从 schema 拉取结果中反序列化 %,d 个时间序列列。"
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 c08dbae23a0..311bf04322a 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
@@ -48,6 +48,7 @@ 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.planner.memory.OperatorMemoryNotEnoughException;
import
org.apache.iotdb.db.queryengine.plan.relational.function.tvf.read_tsfile.ExternalTsFileQueryResource;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.spill.DeviceEntryIOContext;
import
org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ExplainOutputFormat;
@@ -741,7 +742,7 @@ public class MPPQueryContext implements IAuditEntity {
schemaFetchDeserializedColumnCount = 0;
}
- private MemoryNotEnoughException
enrichResultSetColumnMemoryNotEnoughException(
+ MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException(
MemoryNotEnoughException e, long requestedBytes) {
if (!resultSetColumnMemoryTrackingEnabled
|| (matchedSourceColumnsForResultSet == 0
@@ -750,10 +751,23 @@ public class MPPQueryContext implements IAuditEntity {
return e;
}
- long freeBytes =
LocalExecutionPlanner.getInstance().getFreeMemoryForOperators();
+ long freeBytes =
+ e instanceof OperatorMemoryNotEnoughException
+ ? ((OperatorMemoryNotEnoughException) e).getFreeBytes()
+ : LocalExecutionPlanner.getInstance().getFreeMemoryForOperators();
+ long failedReservationBytes =
+ e instanceof OperatorMemoryNotEnoughException
+ ? ((OperatorMemoryNotEnoughException) e).getRequestedBytes()
+ : requestedBytes;
long shortageBytes =
- requestedBytes > 0 && requestedBytes > freeBytes ? requestedBytes -
freeBytes : -1;
+ failedReservationBytes > 0 && failedReservationBytes > freeBytes
+ ? failedReservationBytes - freeBytes
+ : -1;
long exceededColumns = estimateExceededColumns(freeBytes, requestedBytes);
+ long columnEquivalentShortage =
+ exceededColumns > 0
+ ? 0
+ : estimateColumnEquivalentShortage(freeBytes,
failedReservationBytes, requestedBytes);
return new MemoryNotEnoughException(
String.format(
@@ -767,7 +781,10 @@ public class MPPQueryContext implements IAuditEntity {
Locale.ROOT,
DataNodeQueryMessages.RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY,
exceededColumns)
- : "",
+ : String.format(
+ Locale.ROOT,
+
DataNodeQueryMessages.RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT,
+ columnEquivalentShortage),
formatSeriesPaginationForDiagnostics(),
alignByDeviceForResultSetColumnTracking
? ""
@@ -780,7 +797,7 @@ public class MPPQueryContext implements IAuditEntity {
: DataNodeQueryMessages.FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL,
formatBytes(sourceColumnMemoryCostForResultSet),
formatBytes(generatedResultSetColumnMemoryCost),
- formatBytes(requestedBytes),
+ formatBytes(failedReservationBytes),
formatBytes(freeBytes),
e.getMessage()));
}
@@ -841,6 +858,26 @@ public class MPPQueryContext implements IAuditEntity {
return Math.max(0, columnsToCompare - estimatedCapacity);
}
+ /** Converts a failed batch's memory deficit into an observed-column-size
equivalent. */
+ private long estimateColumnEquivalentShortage(
+ long freeBytes, long failedReservationBytes, long requestedBytes) {
+ long avgColumnMemory;
+ if (generatedResultSetColumns > 0 && generatedResultSetColumnMemoryCost >
0) {
+ avgColumnMemory =
+ Math.max(1, divideCeil(generatedResultSetColumnMemoryCost,
generatedResultSetColumns));
+ } else if (expandedSourceColumnsForResultSet > 0 &&
sourceColumnMemoryCostForResultSet > 0) {
+ avgColumnMemory =
+ Math.max(
+ 1, divideCeil(sourceColumnMemoryCostForResultSet,
expandedSourceColumnsForResultSet));
+ } else {
+ avgColumnMemory = Math.max(1, requestedBytes > 0 ? requestedBytes :
failedReservationBytes);
+ }
+
+ // The failed allocation proves a shortage, even if memory was released
before it was read.
+ long shortageBytes = Math.max(1, failedReservationBytes - freeBytes);
+ return divideCeil(shortageBytes, avgColumnMemory);
+ }
+
private long estimateExceededSchemaFetchColumns(long freeBytes, long
requestedBytes) {
if (schemaFetchDeserializedColumnCount <= 0) {
return -1;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
index 791558158ff..feb6b0cb74a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlanner.java
@@ -40,6 +40,7 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContex
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.queryengine.metric.QueryRelatedResourceMetricSet;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
+import
org.apache.iotdb.db.queryengine.plan.planner.memory.OperatorMemoryNotEnoughException;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
import
org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl;
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
@@ -339,15 +340,18 @@ public class LocalExecutionPlanner {
}
long allocated = allocateOperatorsMemory(memoryInBytes, isHighestPriority);
if (allocated < 0) {
- throw new MemoryNotEnoughException(
+ long freeBytes = OPERATORS_MEMORY_BLOCK.getFreeMemoryInBytes();
+ throw new OperatorMemoryNotEnoughException(
String.format(
DataNodeQueryMessages
.QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02,
queryId,
contextHolder,
- OPERATORS_MEMORY_BLOCK.getFreeMemoryInBytes(),
+ freeBytes,
reservedBytes,
- memoryInBytes));
+ memoryInBytes),
+ memoryInBytes,
+ freeBytes);
}
return allocated;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/OperatorMemoryNotEnoughException.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/OperatorMemoryNotEnoughException.java
new file mode 100644
index 00000000000..fb77b88a571
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/OperatorMemoryNotEnoughException.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.planner.memory;
+
+import org.apache.iotdb.calc.exception.MemoryNotEnoughException;
+
+/** Preserves the failed batch size and available memory for query-analysis
diagnostics. */
+public class OperatorMemoryNotEnoughException extends MemoryNotEnoughException
{
+
+ private final long requestedBytes;
+ private final long freeBytes;
+
+ public OperatorMemoryNotEnoughException(String message, long requestedBytes,
long freeBytes) {
+ super(message);
+ this.requestedBytes = requestedBytes;
+ this.freeBytes = freeBytes;
+ }
+
+ public long getRequestedBytes() {
+ return requestedBytes;
+ }
+
+ public long getFreeBytes() {
+ return freeBytes;
+ }
+}
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
index 9f5803e1930..fca6a3600cd 100644
---
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
@@ -21,8 +21,11 @@ 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.apache.iotdb.db.queryengine.plan.planner.memory.OperatorMemoryNotEnoughException;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Test;
public class MPPQueryContextTest {
@@ -60,6 +63,61 @@ public class MPPQueryContextTest {
assertContains(message, "Original error:");
}
+ @Test
+ public void
resultSetColumnMemoryNotEnoughExceptionReportsOverageAfterExpansion() {
+ // The old estimate returned zero when all (or all but one) source columns
were expanded.
+
Assume.assumeTrue(LocalExecutionPlanner.getInstance().getFreeMemoryForOperators()
> 0);
+ for (int unmatchedColumns = 0; unmatchedColumns <= 1; unmatchedColumns++) {
+ MPPQueryContext context =
+ new MPPQueryContext(new QueryId("result_column_overage_" +
unmatchedColumns));
+ context.initResultSetColumnMemoryTracking(0, 0, false);
+ context.recordMatchedSourceColumnsForResultSet(2 + unmatchedColumns);
+ context.recordExpandedSourceColumnForResultSet(1);
+ context.recordExpandedSourceColumnForResultSet(1);
+ context.recordGeneratedResultSetColumn(2);
+
+ MemoryNotEnoughException exception =
+ Assert.assertThrows(
+ MemoryNotEnoughException.class,
+ () ->
context.reserveMemoryForFrontEnd(requestLargerThanFreeOperatorMemory()));
+
+ String message = exception.getMessage();
+ assertContains(message, "expanded 2 source columns, and generated 1
result-set columns");
+ assertContains(message, "equivalent of at least 1 columns");
+ Assert.assertFalse(message, message.contains("The matched source columns
exceed"));
+ assertContains(message, "increase query memory by at least");
+ }
+ }
+
+ @Test
+ public void resultSetColumnMemoryNotEnoughExceptionUsesFailedBatchSize() {
+ MPPQueryContext context = new MPPQueryContext(new
QueryId("result_column_batch_oom_test"));
+ context.initResultSetColumnMemoryTracking(0, 0, false);
+ context.recordMatchedSourceColumnsForResultSet(2);
+ context.recordExpandedSourceColumnForResultSet(1);
+ context.recordExpandedSourceColumnForResultSet(1);
+ context.recordGeneratedResultSetColumn(2);
+
+ long failedBatchBytes = 1_048_840;
+ long freeBytesAtFailure = 702_452;
+ long lastExpressionBytes = 760;
+ MemoryNotEnoughException original =
+ new OperatorMemoryNotEnoughException(
+ "the memory requested this time is 1048840B", failedBatchBytes,
freeBytesAtFailure);
+
+ MemoryNotEnoughException exception =
+ context.enrichResultSetColumnMemoryNotEnoughException(original,
lastExpressionBytes);
+ String message = exception.getMessage();
+
+ Assert.assertEquals(
+ TSStatusCode.QUERY_EXECUTION_MEMORY_NOT_ENOUGH.getStatusCode(),
exception.getErrorCode());
+ assertContains(message, "requested this time 1.00 MB (1048840 B)");
+ assertContains(message, "increase query memory by at least 338.27 KB
(346388 B)");
+ assertContains(message, "at least 173,194 columns");
+ assertContains(message, "the memory requested this time is 1048840B");
+ Assert.assertFalse(message, message.contains("requested this time 760 B"));
+ }
+
@Test
public void
schemaFetchMemoryNotEnoughExceptionContainsFetchedColumnDiagnostics() {
MPPQueryContext context = new MPPQueryContext(new
QueryId("schema_fetch_oom_test"));
