Fixed the updateAndGet side effect in TableDeviceCacheEntry and PipeRemainingOperator

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 ////////////////////////////

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) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java

@@ -38,7 +38,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.OptionalLong;
-import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
 import static org.apache.iotdb.commons.schema.SchemaConstant.NON_TEMPLATE;
@@ -80,29 +79,21 @@ int updateAttribute(
   }
 
   int invalidateAttribute() {
-    final AtomicInteger size = new AtomicInteger(0);
-    deviceSchema.updateAndGet(
-        schema -> {
-          if (schema instanceof TableAttributeSchema) {
-            size.set(schema.estimateSize());
-            return null;
-          }
-          return schema;
-        });
-    return size.get();
+    IDeviceSchema schema;
+    do {
+      schema = deviceSchema.get();
+      if (!(schema instanceof TableAttributeSchema)) {
+        return 0;
+      }
+    } while (!deviceSchema.compareAndSet(schema, null));
+    return schema.estimateSize();
   }
 
   int invalidateAttributeColumn(final String attribute) {
-    final AtomicInteger size = new AtomicInteger(0);
-    deviceSchema.updateAndGet(
-        schema -> {
-          if (schema instanceof TableAttributeSchema) {
-            size.set(((TableAttributeSchema) schema).removeAttribute(attribute));
-            return schema;
-          }
-          return schema;
-        });
-    return size.get();
+    final IDeviceSchema schema = deviceSchema.get();
+    return schema instanceof TableAttributeSchema
+        ? ((TableAttributeSchema) schema).removeAttribute(attribute)
+        : 0;
   }
 
   Map<String, Binary> getAttributeMap() {
@@ -162,17 +153,15 @@ IDeviceSchema getDeviceSchema() {
   }
 
   int invalidateTreeSchema() {
-    final AtomicInteger size = new AtomicInteger(0);
-    deviceSchema.updateAndGet(
-        schema -> {
-          if (schema instanceof TreeDeviceNormalSchema
-              || schema instanceof TreeDeviceTemplateSchema) {
-            size.set(schema.estimateSize());
-            return null;
-          }
-          return schema;
-        });
-    return size.get();
+    IDeviceSchema schema;
+    do {
+      schema = deviceSchema.get();
+      if (!(schema instanceof TreeDeviceNormalSchema)
+          && !(schema instanceof TreeDeviceTemplateSchema)) {
+        return 0;
+      }
+    } while (!deviceSchema.compareAndSet(schema, null));
+    return schema.estimateSize();
   }
 
   /////////////////////////////// Last Cache ///////////////////////////////
@@ -240,15 +229,14 @@ Optional<Pair<OptionalLong, TsPrimitiveType[]>> getLastRow(
   }
 
   int invalidateLastCache() {
-    final AtomicInteger size = new AtomicInteger(0);
-    lastCache.updateAndGet(
-        cacheEntry -> {
-          if (Objects.nonNull(cacheEntry)) {
-            size.set(cacheEntry.estimateSize());
-          }
-          return null;
-        });
-    return size.get();
+    TableDeviceLastCache cacheEntry;
+    do {
+      cacheEntry = lastCache.get();
+      if (Objects.isNull(cacheEntry)) {
+        return 0;
+      }
+    } while (!lastCache.compareAndSet(cacheEntry, null));
+    return cacheEntry.estimateSize();
   }
 
   /////////////////////////////// Management ///////////////////////////////