From 9ce3872423ad7e5ab1960d6d40d6cbe9262f4339 Mon Sep 17 00:00:00 2001
From: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date: Thu, 30 Apr 2026 15:00:04 +0800
Subject: [PATCH 1/2] Fixed the updateAndGet side effect in
 TableDeviceCacheEntry and PipeRemainingOperator

---
 .../PipeConfigNodeRemainingTimeOperator.java  | 24 ++++------
 ...DataNodeRemainingEventAndTimeOperator.java | 48 +++++++------------
 2 files changed, 27 insertions(+), 45 deletions(-)

diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeOperator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeOperator.java
index c27e4cd898b06..ff9a7ea2648d5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeOperator.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/metric/overview/PipeConfigNodeRemainingTimeOperator.java
@@ -66,14 +66,11 @@ class PipeConfigNodeRemainingTimeOperator extends PipeRemainingOperator {
             .reduce(Long::sum)
             .orElse(0L);
 
-    configRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            lastConfigRegionCommitSmoothingValue =
-                pipeRemainingTimeCommitRateAverageTime.getMeterRate(meter);
-          }
-          return meter;
-        });
+    final Meter configRegionMeter = configRegionCommitMeter.get();
+    if (Objects.nonNull(configRegionMeter)) {
+      lastConfigRegionCommitSmoothingValue =
+          pipeRemainingTimeCommitRateAverageTime.getMeterRate(configRegionMeter);
+    }
 
     final double configRegionRemainingTime;
     if (totalConfigRegionWriteEventCount <= 0) {
@@ -101,13 +98,10 @@ void register(final IoTDBConfigRegionSource extractor) {
   //////////////////////////// Rate ////////////////////////////
 
   void markConfigRegionCommit() {
-    configRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            meter.mark();
-          }
-          return meter;
-        });
+    final Meter meter = configRegionCommitMeter.get();
+    if (Objects.nonNull(meter)) {
+      meter.mark();
+    }
   }
 
   //////////////////////////// Switch ////////////////////////////
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
index 38cfb579622a5..6ff3de4d9495c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
@@ -138,14 +138,11 @@ public double getRemainingTime() {
             + rawTabletEventCount.get()
             + insertNodeEventCount.get();
 
-    dataRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            lastDataRegionCommitSmoothingValue =
-                pipeRemainingTimeCommitRateAverageTime.getMeterRate(meter);
-          }
-          return meter;
-        });
+    final Meter dataRegionMeter = dataRegionCommitMeter.get();
+    if (Objects.nonNull(dataRegionMeter)) {
+      lastDataRegionCommitSmoothingValue =
+          pipeRemainingTimeCommitRateAverageTime.getMeterRate(dataRegionMeter);
+    }
     final double dataRegionRemainingTime;
     if (totalDataRegionWriteEventCount <= 0) {
       dataRegionRemainingTime = 0;
@@ -162,14 +159,11 @@ public double getRemainingTime() {
             .reduce(Long::sum)
             .orElse(0L);
 
-    schemaRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            lastSchemaRegionCommitSmoothingValue =
-                pipeRemainingTimeCommitRateAverageTime.getMeterRate(meter);
-          }
-          return meter;
-        });
+    final Meter schemaRegionMeter = schemaRegionCommitMeter.get();
+    if (Objects.nonNull(schemaRegionMeter)) {
+      lastSchemaRegionCommitSmoothingValue =
+          pipeRemainingTimeCommitRateAverageTime.getMeterRate(schemaRegionMeter);
+    }
     final double schemaRegionRemainingTime;
     if (totalSchemaRegionWriteEventCount <= 0) {
       schemaRegionRemainingTime = 0;
@@ -199,23 +193,17 @@ void register(final IoTDBSchemaRegionSource source) {
   //////////////////////////// Rate ////////////////////////////
 
   void markDataRegionCommit() {
-    dataRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            meter.mark();
-          }
-          return meter;
-        });
+    final Meter meter = dataRegionCommitMeter.get();
+    if (Objects.nonNull(meter)) {
+      meter.mark();
+    }
   }
 
   void markSchemaRegionCommit() {
-    schemaRegionCommitMeter.updateAndGet(
-        meter -> {
-          if (Objects.nonNull(meter)) {
-            meter.mark();
-          }
-          return meter;
-        });
+    final Meter meter = schemaRegionCommitMeter.get();
+    if (Objects.nonNull(meter)) {
+      meter.mark();
+    }
   }
 
   void markTsFileCollectInvocationCount(final long collectInvocationCount) {

From 72b3155bacb1c19017d71b896075f57148f827ab Mon Sep 17 00:00:00 2001
From: Caideyipi <87789683+Caideyipi@users.noreply.github.com>
Date: Wed, 6 May 2026 14:37:46 +0800
Subject: [PATCH 2/2] Update DeviceCacheEntry.java

---
 .../cache/schema/DeviceCacheEntry.java        | 35 +++++++++----------
 1 file changed, 16 insertions(+), 19 deletions(-)

diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DeviceCacheEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DeviceCacheEntry.java
index 3337e2f1420e9..ecd4c297e5729 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DeviceCacheEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/cache/schema/DeviceCacheEntry.java
@@ -30,7 +30,6 @@
 
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.apache.iotdb.commons.schema.SchemaConstant.NON_TEMPLATE;
@@ -94,15 +93,14 @@ IDeviceSchema getDeviceSchema() {
   }
 
   int invalidateSchema() {
-    final AtomicInteger size = new AtomicInteger(0);
-    deviceSchema.updateAndGet(
-        schema -> {
-          if (Objects.nonNull(schema)) {
-            size.set(schema.estimateSize());
-          }
-          return null;
-        });
-    return size.get();
+    IDeviceSchema schema;
+    do {
+      schema = deviceSchema.get();
+      if (Objects.isNull(schema)) {
+        return 0;
+      }
+    } while (!deviceSchema.compareAndSet(schema, null));
+    return schema.estimateSize();
   }
 
   /////////////////////////////// Last Cache ///////////////////////////////
@@ -150,15 +148,14 @@ boolean updateInputMap(final @Nonnull Map<String, TimeValuePair> updateMap) {
   }
 
   int invalidateLastCache() {
-    final AtomicInteger size = new AtomicInteger(0);
-    lastCache.updateAndGet(
-        cacheEntry -> {
-          if (Objects.nonNull(cacheEntry)) {
-            size.set(cacheEntry.estimateSize());
-          }
-          return null;
-        });
-    return size.get();
+    DeviceLastCache cacheEntry;
+    do {
+      cacheEntry = lastCache.get();
+      if (Objects.isNull(cacheEntry)) {
+        return 0;
+      }
+    } while (!lastCache.compareAndSet(cacheEntry, null));
+    return cacheEntry.estimateSize();
   }
 
   /////////////////////////////// Management ///////////////////////////////