diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index f0c2d609fc3fc..ef9e3a037271f 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -29,6 +29,7 @@ iotdb-server IoTDB: Core: Data-Node (Server) + en ${iotdb.test.skip} false ${iotdb.test.skip} @@ -372,6 +373,19 @@ + + com.diffplug.spotless + spotless-maven-plugin + + + + src/main/java/**/*.java + src/test/java/**/*.java + src/main/i18n/**/*.java + + + + @@ -499,6 +513,7 @@ ${project.build.directory}/generated-sources/freemarker + ${project.basedir}/src/main/i18n/${i18n.locale} @@ -536,6 +551,12 @@ + + with-zh-locale + + zh + + skipIoTDBTests 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 0000000000000..1c4433d3b552e --- /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 0000000000000..5b151ce3e351b --- /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 e2103817b24b1..73d4093e6c7d6 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 @@ 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.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 @@ private void resetResultSetColumnMemoryTracking() { schemaFetchDeserializedColumnCount = 0; } - private MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException( + MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException( MemoryNotEnoughException e, long requestedBytes) { if (!resultSetColumnMemoryTrackingEnabled || (matchedSourceColumnsForResultSet == 0 @@ -527,10 +529,23 @@ private MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException( 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 @@ private MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException( 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 @@ private MemoryNotEnoughException enrichResultSetColumnMemoryNotEnoughException( : FOR_QUERY_ENGINE_OPERATOR_MEMORY_POOL, formatBytes(sourceColumnMemoryCostForResultSet), formatBytes(generatedResultSetColumnMemoryCost), - formatBytes(requestedBytes), + formatBytes(failedReservationBytes), formatBytes(freeBytes), e.getMessage())); } @@ -608,6 +626,26 @@ private long estimateExceededColumns(long freeBytes, long requestedBytes) { 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 b8aef336e1010..04dc7099539da 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.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.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 long reserveFromFreeMemoryForOperators( } 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 0000000000000..8e8c6749559d7 --- /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 2d717f0fc65e1..6d5cd68e04989 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 @@ 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 void resultSetColumnMemoryNotEnoughExceptionContainsColumnDiagnostics() { 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"));