This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 414aa9eff66 [Query] Fix result-column OOM diagnostics (#18661) (#18667)
414aa9eff66 is described below
commit 414aa9eff66310c389764ec1640ee4aac8512155
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 21 11:48:13 2026 +0800
[Query] Fix result-column OOM diagnostics (#18661) (#18667)
(cherry picked from commit 25bf38eba0cc9d3d213b79068ea002e62c5ced96)
---
iotdb-core/datanode/pom.xml | 21 ++++++++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 35 +++++++++++++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 32 ++++++++++++
.../db/queryengine/common/MPPQueryContext.java | 48 +++++++++++++++--
.../plan/planner/LocalExecutionPlanner.java | 19 ++++---
.../memory/OperatorMemoryNotEnoughException.java | 43 ++++++++++++++++
.../db/queryengine/common/MPPQueryContextTest.java | 60 ++++++++++++++++++++++
7 files changed, 247 insertions(+), 11 deletions(-)
diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
index f0c2d609fc3..ef9e3a03727 100644
--- a/iotdb-core/datanode/pom.xml
+++ b/iotdb-core/datanode/pom.xml
@@ -29,6 +29,7 @@
<artifactId>iotdb-server</artifactId>
<name>IoTDB: Core: Data-Node (Server)</name>
<properties>
+ <i18n.locale>en</i18n.locale>
<iotdb.it.skip>${iotdb.test.skip}</iotdb.it.skip>
<iotdb.test.skip>false</iotdb.test.skip>
<iotdb.ut.skip>${iotdb.test.skip}</iotdb.ut.skip>
@@ -372,6 +373,19 @@
</dependencies>
<build>
<plugins>
+ <plugin>
+ <groupId>com.diffplug.spotless</groupId>
+ <artifactId>spotless-maven-plugin</artifactId>
+ <configuration>
+ <java>
+ <includes>
+ <include>src/main/java/**/*.java</include>
+ <include>src/test/java/**/*.java</include>
+ <include>src/main/i18n/**/*.java</include>
+ </includes>
+ </java>
+ </configuration>
+ </plugin>
<!--using `mvn test` to run UT, `mvn verify` to run ITs
Reference:
https://antoniogoncalves.org/2012/12/13/lets-turn-integration-tests-with-maven-to-a-first-class-citizen/-->
<plugin>
@@ -499,6 +513,7 @@
<configuration>
<sources>
<source>${project.build.directory}/generated-sources/freemarker</source>
+
<source>${project.basedir}/src/main/i18n/${i18n.locale}</source>
</sources>
</configuration>
</execution>
@@ -536,6 +551,12 @@
</plugins>
</build>
<profiles>
+ <profile>
+ <id>with-zh-locale</id>
+ <properties>
+ <i18n.locale>zh</i18n.locale>
+ </properties>
+ </profile>
<profile>
<id>skipIoTDBTests</id>
<activation>
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
new file mode 100644
index 00000000000..1c4433d3b55
--- /dev/null
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -0,0 +1,35 @@
+/*
+ * 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.i18n;
+
+public final class DataNodeQueryMessages {
+
+ private DataNodeQueryMessages() {}
+
+ 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
+
QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02
=
+ "There is not enough memory for Query %s, the contextHolder is
%s,current remaining free "
+ + "memory is %dB, already reserved memory for this context in
total is %dB, the memory "
+ + "requested this time is %dB";
+}
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
new file mode 100644
index 00000000000..5b151ce3e35
--- /dev/null
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -0,0 +1,32 @@
+/*
+ * 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.i18n;
+
+public final class DataNodeQueryMessages {
+
+ private DataNodeQueryMessages() {}
+
+ public static final String RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT =
+ "本次失败的内存申请超出可用内存，按已记录列的平均内存估算，至少超出相当于 %,d 列的容量。";
+
+ public static final String
+
QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02
=
+ "Query %s 内存不足，contextHolder 为 %s，当前剩余空闲内存为 %dB，该 context 已预留总内存为
%dB，本次请求内存为 %dB。";
+}
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 e2103817b24..73d4093e6c7 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
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.common;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
import org.apache.iotdb.commons.utils.TestOnly;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.exception.MemoryNotEnoughException;
import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
import org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils;
@@ -31,6 +32,7 @@ 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.MemoryReservationManager;
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.statistics.QueryPlanStatistics;
import org.apache.tsfile.read.filter.basic.Filter;
@@ -518,7 +520,7 @@ public class MPPQueryContext {
schemaFetchDeserializedColumnCount = 0;
}
- private MemoryNotEnoughException
enrichResultSetColumnMemoryNotEnoughException(
+ MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException(
MemoryNotEnoughException e, long requestedBytes) {
if (!resultSetColumnMemoryTrackingEnabled
|| (matchedSourceColumnsForResultSet == 0
@@ -527,10 +529,23 @@ public class MPPQueryContext {
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(
@@ -542,7 +557,10 @@ public class MPPQueryContext {
exceededColumns > 0
? String.format(
Locale.ROOT, RESULT_SET_COLUMNS_EXCEED_MEMORY_CAPACITY,
exceededColumns)
- : "",
+ : String.format(
+ Locale.ROOT,
+
DataNodeQueryMessages.RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT,
+ columnEquivalentShortage),
formatSeriesPaginationForDiagnostics(),
alignByDeviceForResultSetColumnTracking
? ""
@@ -552,7 +570,7 @@ public class MPPQueryContext {
: FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL,
formatBytes(sourceColumnMemoryCostForResultSet),
formatBytes(generatedResultSetColumnMemoryCost),
- formatBytes(requestedBytes),
+ formatBytes(failedReservationBytes),
formatBytes(freeBytes),
e.getMessage()));
}
@@ -608,6 +626,26 @@ public class MPPQueryContext {
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 b8aef336e10..04dc7099539 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
@@ -22,6 +22,7 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.common.DeviceContext;
import org.apache.iotdb.db.queryengine.exception.MemoryNotEnoughException;
import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
@@ -31,6 +32,7 @@ import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateM
import org.apache.iotdb.db.queryengine.execution.operator.Operator;
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.planner.memory.PipelineMemoryEstimator;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.schemaengine.schemaregion.ISchemaRegion;
@@ -280,13 +282,18 @@ public class LocalExecutionPlanner {
}
long allocated = allocateOperatorsMemory(memoryInBytes, isHighestPriority);
if (allocated < 0) {
- throw new MemoryNotEnoughException(
+ long freeBytes = freeMemoryForOperators;
+ throw new OperatorMemoryNotEnoughException(
String.format(
- "There is not enough memory for Query %s, the contextHolder is
%s,"
- + "current remaining free memory is %dB, "
- + "already reserved memory for this context in total is %dB,
"
- + "the memory requested this time is %dB",
- queryId, contextHolder, freeMemoryForOperators, reservedBytes,
memoryInBytes));
+ DataNodeQueryMessages
+
.QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02,
+ queryId,
+ contextHolder,
+ freeBytes,
+ reservedBytes,
+ 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..8e8c6749559
--- /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.db.queryengine.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 2d717f0fc65..6d5cd68e049 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,12 @@ package org.apache.iotdb.db.queryengine.common;
import org.apache.iotdb.db.queryengine.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.db.utils.ErrorHandlingUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Test;
public class MPPQueryContextTest {
@@ -60,6 +64,62 @@ 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.QUOTA_MEM_QUERY_NOT_ENOUGH.getStatusCode(),
+ ErrorHandlingUtils.onQueryException(exception, "query").getCode());
+ 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"));
