From b58b4b06f3a2a7a9aaf9e32e170bd399d27e7e1a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 15 Sep 2025 17:43:01 +0800 Subject: [PATCH 1/2] Pipe: Further reduced the repeating exception log & Changed the default log reducer size to 16M & Did some minor fixes in Device Schema Cache (#16417) * fix * fix --- .../async/IoTDBDataRegionAsyncSink.java | 9 ++-- .../dualkeycache/IDualKeyCacheUpdating.java | 45 ------------------- .../iotdb/commons/conf/CommonConfig.java | 2 +- 3 files changed, 7 insertions(+), 49 deletions(-) delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java index e9def74eaaa03..2b57d94bfc361 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient; import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.event.EnrichedEvent; +import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink; import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent; import org.apache.iotdb.db.pipe.event.common.schema.PipeSchemaRegionWritePlanEvent; @@ -497,11 +498,13 @@ public void updateLeaderCache(final String deviceId, final TEndPoint endPoint) { private void logOnClientException( final AsyncPipeDataTransferServiceClient client, final Exception e) { if (client == null) { - LOGGER.warn(THRIFT_ERROR_FORMATTER_WITHOUT_ENDPOINT); + PipeLogger.log(LOGGER::warn, THRIFT_ERROR_FORMATTER_WITHOUT_ENDPOINT); } else { client.resetMethodStateIfStopped(); - LOGGER.warn( - String.format(THRIFT_ERROR_FORMATTER_WITH_ENDPOINT, client.getIp(), client.getPort()), e); + PipeLogger.log( + LOGGER::warn, + e, + String.format(THRIFT_ERROR_FORMATTER_WITH_ENDPOINT, client.getIp(), client.getPort())); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java deleted file mode 100644 index 7ebdd566c9eff..0000000000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java +++ /dev/null @@ -1,45 +0,0 @@ -/* - * 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.analyze.cache.schema.dualkeycache; - -/** - * This interfaces defines the behaviour needed be implemented and executed during cache value - * traverse. - * - * @param The first key of target cache values - * @param The second key of one target cache value - * @param The cache value - */ -public interface IDualKeyCacheUpdating { - - /** Return the first key of target cache values. */ - FK getFirstKey(); - - /** Return the second key list of target cache values. */ - SK[] getSecondKeyList(); - - /** - * Compute each target cache value. The index is the second key's position in second key list. The - * value here can be updated. - * - * @return increasing memory usage - */ - int updateValue(int index, V value); -} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java index a8c4c1758a819..24d72a5e415e6 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java @@ -300,7 +300,7 @@ public class CommonConfig { private int pipeReceiverReqDecompressedMaxLengthInBytes = 1073741824; // 1GB private boolean pipeReceiverLoadConversionEnabled = false; private volatile long pipePeriodicalLogMinIntervalSeconds = 60; - private volatile long pipeLoggerCacheMaxSizeInBytes = 10 * MB; + private volatile long pipeLoggerCacheMaxSizeInBytes = 16 * MB; private volatile double pipeMetaReportMaxLogNumPerRound = 0.1; private volatile int pipeMetaReportMaxLogIntervalRounds = 360; From 5bf22ae01ad017ea7068b11f9d0733ba5081a824 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 16 Sep 2025 16:07:58 +0800 Subject: [PATCH 2/2] add-back --- .../dualkeycache/IDualKeyCacheUpdating.java | 45 +++++++++++++++++++ 1 file changed, 45 insertions(+) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java new file mode 100644 index 0000000000000..7ebdd566c9eff --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/dualkeycache/IDualKeyCacheUpdating.java @@ -0,0 +1,45 @@ +/* + * 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.analyze.cache.schema.dualkeycache; + +/** + * This interfaces defines the behaviour needed be implemented and executed during cache value + * traverse. + * + * @param The first key of target cache values + * @param The second key of one target cache value + * @param The cache value + */ +public interface IDualKeyCacheUpdating { + + /** Return the first key of target cache values. */ + FK getFirstKey(); + + /** Return the second key list of target cache values. */ + SK[] getSecondKeyList(); + + /** + * Compute each target cache value. The index is the second key's position in second key list. The + * value here can be updated. + * + * @return increasing memory usage + */ + int updateValue(int index, V value); +}