This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e876ddd023a Fix concurrent device attribute update serialization
counts (#18959)
e876ddd023a is described below
commit e876ddd023a4055c62565c553968181e12b7a756
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 19:47:45 2026 +0800
Fix concurrent device attribute update serialization counts (#18959)
---
.../attribute/update/UpdateClearContainer.java | 2 +-
.../attribute/update/UpdateDetailContainer.java | 7 +-
.../update/UpdateContainerSerializationTest.java | 117 +++++++++++++++++++++
3 files changed, 122 insertions(+), 4 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
index 8ee41c2a574..25984fe7cb0 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateClearContainer.java
@@ -120,7 +120,7 @@ public class UpdateClearContainer implements
UpdateContainer {
++setEntryCount;
outputStream.writeWithLength(tableBytes);
}
- outputStream.rewrite(tableNames.size(), setSizeOffset);
+ outputStream.rewrite(setEntryCount, setSizeOffset);
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
index 2883a26512a..fc2c25de22f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateDetailContainer.java
@@ -278,11 +278,12 @@ public class UpdateDetailContainer implements
UpdateContainer {
outputStream.writeWithLength(valueBytes);
++attributeCount;
}
- outputStream.rewrite(deviceEntry.getValue().size(), attributeOffset);
+ // Concurrent updates may change the map size after its iterator has
been exhausted.
+ outputStream.rewrite(attributeCount, attributeOffset);
}
- outputStream.rewrite(tableEntry.getValue().size(), deviceSizeOffset);
+ outputStream.rewrite(deviceEntryCount, deviceSizeOffset);
}
- outputStream.rewrite(updateMap.size(), mapSizeOffset);
+ outputStream.rewrite(mapEntryCount, mapSizeOffset);
}
@Override
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainerSerializationTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainerSerializationTest.java
new file mode 100644
index 00000000000..db72c6c963e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/attribute/update/UpdateContainerSerializationTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.schemaengine.schemaregion.attribute.update;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class UpdateContainerSerializationTest {
+
+ @Test
+ public void testTableAddedDuringSerialization() throws Exception {
+ assertConcurrentUpdateRoundTrip(0);
+ }
+
+ @Test
+ public void testDeviceAddedDuringSerialization() throws Exception {
+ assertConcurrentUpdateRoundTrip(1);
+ }
+
+ @Test
+ public void testAttributeAddedDuringSerialization() throws Exception {
+ assertConcurrentUpdateRoundTrip(2);
+ }
+
+ private void assertConcurrentUpdateRoundTrip(final int level) throws
Exception {
+ final UpdateDetailContainer container = new UpdateDetailContainer();
+ final AtomicReference<Runnable> concurrentUpdate = new
AtomicReference<>(() -> {});
+ final ConcurrentMap<String, Binary> attributes =
+ new ConcurrentHashMap<String, Binary>() {
+ @Override
+ public Set<Map.Entry<String, Binary>> entrySet() {
+ // A weakly consistent iterator may miss an insertion made after
traversal starts.
+ // Apply it here so the race is reproducible without relying on
thread scheduling.
+ final Set<Map.Entry<String, Binary>> entries = new
HashSet<>(super.entrySet());
+ concurrentUpdate.getAndSet(() -> {}).run();
+ return entries;
+ }
+ };
+ final Binary value = new Binary("value", TSFileConfig.STRING_CHARSET);
+ attributes.put("attribute", value);
+ final ConcurrentMap<List<String>, ConcurrentMap<String, Binary>> devices =
+ new ConcurrentHashMap<>();
+ devices.put(Collections.singletonList("device"), attributes);
+ container.getUpdateMap().put("table", devices);
+ concurrentUpdate.set(
+ () -> {
+ if (level == 0) {
+ container.getUpdateMap().put("new_table", new
ConcurrentHashMap<>());
+ } else if (level == 1) {
+ devices.put(Collections.singletonList("new_device"), new
ConcurrentHashMap<>());
+ } else {
+ attributes.put("new_attribute", value);
+ }
+ });
+
+ final AtomicBoolean hasRemaining = new AtomicBoolean();
+ final byte[] bytes =
+ container.getUpdateContent(new AtomicInteger(Integer.MAX_VALUE),
hasRemaining);
+ assertFalse(hasRemaining.get());
+ // Check the lengths before decoding to avoid allocating from a corrupt
string length.
+ final ByteArrayInputStream input = new ByteArrayInputStream(bytes);
+ assertTrue(ReadWriteIOUtils.readBool(input));
+ assertEquals(1, ReadWriteIOUtils.readInt(input));
+ assertEquals("table", ReadWriteIOUtils.readString(input));
+ assertEquals(1, ReadWriteIOUtils.readInt(input));
+ assertEquals(1, ReadWriteIOUtils.readInt(input));
+ assertEquals("device", ReadWriteIOUtils.readString(input));
+ assertEquals(1, ReadWriteIOUtils.readInt(input));
+
+ final UpdateDetailContainer decoded = new UpdateDetailContainer();
+ final ByteArrayInputStream roundTrip = new ByteArrayInputStream(bytes);
+ assertTrue(ReadWriteIOUtils.readBool(roundTrip));
+ decoded.deserialize(roundTrip);
+ assertEquals(0, roundTrip.available());
+ assertEquals(
+ Collections.singletonMap(
+ "table",
+ Collections.singletonMap(
+ Collections.singletonList("device"),
Collections.singletonMap("attribute", value))),
+ decoded.getUpdateMap());
+ }
+}
