From 09a57102ad85f263685fdd37eb7a065ef6d3f74a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 9 Jun 2026 17:14:38 +0800 Subject: [PATCH 1/4] [to dev/1.3] Fix tree model WAL serialized sizes --- .../plan/node/write/DeleteDataNode.java | 2 +- .../plan/node/write/InsertRowNode.java | 2 +- .../plan/node/write/InsertTabletNode.java | 2 +- .../dataregion/modification/Deletion.java | 10 +++- .../dataregion/wal/buffer/WALInfoEntry.java | 9 +++- .../dataregion/wal/utils/WALWriteUtils.java | 16 ++++-- .../node/write/DeleteDataNodeSerdeTest.java | 28 ++++++++++ .../node/write/InsertRowNodeSerdeTest.java | 12 ++--- .../node/write/InsertTabletNodeSerdeTest.java | 12 ++--- .../dataregion/modification/DeletionTest.java | 52 +++++++++++++++++++ .../dataregion/wal/io/WALFileTest.java | 12 ++++- 11 files changed, 135 insertions(+), 22 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionTest.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java index c4c4159f66dea..3097a51fc50fc 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java @@ -129,7 +129,7 @@ public List getOutputColumnNames() { public int serializedSize() { int size = FIXED_SERIALIZED_SIZE; for (PartialPath path : pathList) { - size += ReadWriteIOUtils.sizeToWrite(path.getFullPath()); + size += WALWriteUtils.sizeToWrite(path.getFullPath()); } return size; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java index 6518385c48a86..41c4fbeca9be2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java @@ -524,7 +524,7 @@ public int serializedSize() { protected int subSerializeSize() { int size = 0; size += Long.BYTES; - size += ReadWriteIOUtils.sizeToWrite(devicePath.getFullPath()); + size += WALWriteUtils.sizeToWrite(devicePath.getFullPath()); return size + serializeMeasurementsAndValuesSize(); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index 6a1e9bf1bb48c..283dba678e7e3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java @@ -738,7 +738,7 @@ public int serializedSize(int start, int end) { int subSerializeSize(int start, int end) { int size = 0; size += Long.BYTES; - size += ReadWriteIOUtils.sizeToWrite(devicePath.getFullPath()); + size += WALWriteUtils.sizeToWrite(devicePath.getFullPath()); // measurements size size += Integer.BYTES; size += serializeMeasurementSchemasSize(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java index 07f8aecbfc026..1cb70a4c9cb3f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java @@ -24,6 +24,7 @@ import org.apache.iotdb.commons.utils.PathUtils; import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.utils.RamUsageEstimator; @@ -137,7 +138,14 @@ private static PartialPath getMeasurementPath(String path) throws IllegalPathExc } public long getSerializedSize() { - return Long.BYTES * 2 + Integer.BYTES + (long) getPathString().length() * Character.BYTES; + return Long.BYTES * 2L + sizeToWriteString(getPathString()); + } + + private static int sizeToWriteString(String value) { + if (value == null) { + return Integer.BYTES; + } + return Integer.BYTES + value.getBytes(TSFileConfig.STRING_CHARSET).length; } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java index 15c8faa8993d7..a9af8f81c7d62 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java @@ -64,7 +64,14 @@ public WALInfoEntry(long memTableId, InsertTabletNode value, int tabletStart, in @Override public int serializedSize() { - return FIXED_SERIALIZED_SIZE + (value == null ? 0 : value.serializedSize()); + if (value == null) { + return FIXED_SERIALIZED_SIZE; + } + if (value instanceof InsertTabletNode && tabletInfo != null) { + return FIXED_SERIALIZED_SIZE + + ((InsertTabletNode) value).serializedSize(tabletInfo.tabletStart, tabletInfo.tabletEnd); + } + return FIXED_SERIALIZED_SIZE + value.serializedSize(); } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALWriteUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALWriteUtils.java index e6c0eb02722ae..72dd2671fa313 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALWriteUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALWriteUtils.java @@ -21,6 +21,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.enums.CompressionType; @@ -127,13 +128,20 @@ public static int write(String s, IWALByteBufferView buffer) { return write(NO_BYTE_TO_READ, buffer); } int len = 0; - byte[] bytes = s.getBytes(); + byte[] bytes = s.getBytes(TSFileConfig.STRING_CHARSET); len += write(bytes.length, buffer); buffer.put(bytes); len += bytes.length; return len; } + public static int sizeToWrite(String s) { + if (s == null) { + return INT_LEN; + } + return INT_LEN + s.getBytes(TSFileConfig.STRING_CHARSET).length; + } + /** * Write IDeviceID to byteBuffer. * @@ -196,15 +204,15 @@ public static int write(MeasurementSchema measurementSchema, IWALByteBufferView public static int sizeToWrite(MeasurementSchema measurementSchema) { int byteLen = 0; - byteLen += ReadWriteIOUtils.sizeToWrite(measurementSchema.getMeasurementId()); + byteLen += sizeToWrite(measurementSchema.getMeasurementId()); byteLen += 3 * Byte.BYTES; Map props = measurementSchema.getProps(); byteLen += Integer.BYTES; if (props != null) { for (Map.Entry entry : props.entrySet()) { - byteLen += ReadWriteIOUtils.sizeToWrite(entry.getKey()); - byteLen += ReadWriteIOUtils.sizeToWrite(entry.getValue()); + byteLen += sizeToWrite(entry.getKey()); + byteLen += sizeToWrite(entry.getValue()); } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/DeleteDataNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/DeleteDataNodeSerdeTest.java index cbfba2a687914..3087085538326 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/DeleteDataNodeSerdeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/DeleteDataNodeSerdeTest.java @@ -25,10 +25,14 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.junit.Assert; import org.junit.Test; +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; @@ -64,4 +68,28 @@ public void testSerializeAndDeserialize() throws IllegalPathException { Assert.assertEquals(pathList.get(i), deserializedPathList.get(i)); } } + + @Test + public void testSerializeAndDeserializeForWAL() throws IllegalPathException, IOException { + long startTime = 1; + long endTime = 10; + List pathList = new ArrayList<>(); + pathList.add(new PartialPath("root.\u6570\u636e\u5e93.d1.\u6e29\u5ea6")); + pathList.add(new PartialPath("root.\u6570\u636e\u5e93.d2.*")); + DeleteDataNode deleteDataNode = + new DeleteDataNode(new PlanNodeId("DeleteDataNode"), pathList, startTime, endTime); + + ByteBuffer byteBuffer = ByteBuffer.allocate(deleteDataNode.serializedSize()); + deleteDataNode.serializeToWAL(new WALByteBufferForTest(byteBuffer)); + Assert.assertEquals(deleteDataNode.serializedSize(), byteBuffer.position()); + + DataInputStream dataInputStream = + new DataInputStream(new ByteArrayInputStream(byteBuffer.array())); + Assert.assertEquals(PlanNodeType.DELETE_DATA.getNodeType(), dataInputStream.readShort()); + + DeleteDataNode deserializedNode = DeleteDataNode.deserializeFromWAL(dataInputStream); + Assert.assertEquals(startTime, deserializedNode.getDeleteStartTime()); + Assert.assertEquals(endTime, deserializedNode.getDeleteEndTime()); + Assert.assertEquals(pathList, deserializedNode.getPathList()); + } } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java index 523b061428241..a4e19d9717629 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertRowNodeSerdeTest.java @@ -89,8 +89,8 @@ public void TestSerializeAndDeserializeForWAL() throws IllegalPathException, IOE tmpNode.setPlanNodeId(insertRowNode.getPlanNodeId()); tmpNode.setMeasurementSchemas( new MeasurementSchema[] { - new MeasurementSchema("s1", TSDataType.DOUBLE), - new MeasurementSchema("s2", TSDataType.FLOAT), + new MeasurementSchema("\u6e29\u5ea6", TSDataType.DOUBLE), + new MeasurementSchema("\u6e7f\u5ea6", TSDataType.FLOAT), new MeasurementSchema("s3", TSDataType.INT64), new MeasurementSchema("s4", TSDataType.INT32), new MeasurementSchema("s5", TSDataType.BOOLEAN) @@ -148,9 +148,9 @@ private InsertRowNode getInsertRowNodeWithMeasurementSchemas() throws IllegalPat InsertRowNode insertRowNode = new InsertRowNode( new PlanNodeId("plannode 2"), - new PartialPath("root.isp.d2"), + new PartialPath("root.\u6570\u636e\u5e93.d2"), false, - new String[] {"s1", "s2", "s3", "s4", "s5"}, + new String[] {"\u6e29\u5ea6", "\u6e7f\u5ea6", "s3", "s4", "s5"}, dataTypes, time, columns, @@ -158,8 +158,8 @@ private InsertRowNode getInsertRowNodeWithMeasurementSchemas() throws IllegalPat insertRowNode.setMeasurementSchemas( new MeasurementSchema[] { - new MeasurementSchema("s1", TSDataType.DOUBLE), - new MeasurementSchema("s2", TSDataType.FLOAT), + new MeasurementSchema("\u6e29\u5ea6", TSDataType.DOUBLE), + new MeasurementSchema("\u6e7f\u5ea6", TSDataType.FLOAT), new MeasurementSchema("s3", TSDataType.INT64), new MeasurementSchema("s4", TSDataType.INT32), new MeasurementSchema("s5", TSDataType.BOOLEAN) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java index 754e7a6e1386d..a883b7d24a87e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertTabletNodeSerdeTest.java @@ -84,8 +84,8 @@ public void testSerializeAndDeserializeForWAL() throws IllegalPathException, IOE tmpNode.setMeasurementSchemas( new MeasurementSchema[] { - new MeasurementSchema("s1", TSDataType.DOUBLE), - new MeasurementSchema("s2", TSDataType.FLOAT), + new MeasurementSchema("\u6e29\u5ea6", TSDataType.DOUBLE), + new MeasurementSchema("\u6e7f\u5ea6", TSDataType.FLOAT), new MeasurementSchema("s3", TSDataType.INT64), new MeasurementSchema("s4", TSDataType.INT32), new MeasurementSchema("s5", TSDataType.BOOLEAN) @@ -191,9 +191,9 @@ private InsertTabletNode getInsertTabletNodeWithSchema() throws IllegalPathExcep InsertTabletNode insertTabletNode = new InsertTabletNode( new PlanNodeId("plannode 1"), - new PartialPath("root.isp.d1"), + new PartialPath("root.\u6570\u636e\u5e93.d1"), false, - new String[] {"s1", "s2", "s3", "s4", "s5"}, + new String[] {"\u6e29\u5ea6", "\u6e7f\u5ea6", "s3", "s4", "s5"}, dataTypes, times, null, @@ -201,8 +201,8 @@ private InsertTabletNode getInsertTabletNodeWithSchema() throws IllegalPathExcep times.length); insertTabletNode.setMeasurementSchemas( new MeasurementSchema[] { - new MeasurementSchema("s1", TSDataType.DOUBLE), - new MeasurementSchema("s2", TSDataType.FLOAT), + new MeasurementSchema("\u6e29\u5ea6", TSDataType.DOUBLE), + new MeasurementSchema("\u6e7f\u5ea6", TSDataType.FLOAT), new MeasurementSchema("s3", TSDataType.INT64), new MeasurementSchema("s4", TSDataType.INT32), new MeasurementSchema("s5", TSDataType.BOOLEAN) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionTest.java new file mode 100644 index 0000000000000..c8c1a7fd0ca8d --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/modification/DeletionTest.java @@ -0,0 +1,52 @@ +/* + * 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.storageengine.dataregion.modification; + +import org.apache.iotdb.commons.path.PartialPath; + +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataInputStream; +import java.io.DataOutputStream; + +import static org.junit.Assert.assertEquals; + +public class DeletionTest { + + @Test + public void testSerializedSize() throws Exception { + Deletion deletion = + new Deletion(new PartialPath("root.\u6570\u636e\u5e93.d1.\u6e29\u5ea6"), 0, 1, 5); + + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); + long serializedSize = + deletion.serializeWithoutFileOffset(new DataOutputStream(byteArrayOutputStream)); + byte[] bytes = byteArrayOutputStream.toByteArray(); + + assertEquals(deletion.getSerializedSize(), serializedSize); + assertEquals(deletion.getSerializedSize(), bytes.length); + assertEquals( + deletion, + Deletion.deserializeWithoutFileOffset( + new DataInputStream(new ByteArrayInputStream(bytes)))); + } +} diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java index f483bcfd49265..8e2f45ce5564c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileTest.java @@ -61,7 +61,7 @@ public class WALFileTest { new File( TestConstant.BASE_OUTPUT_PATH.concat( WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))); - private final String devicePath = "root.test_sg.test_d"; + private final String devicePath = "root.\u6570\u636e\u5e93.test_d"; @Before public void setUp() throws Exception { @@ -171,6 +171,16 @@ public void testReadMetadataFromBrokenFile() throws IOException { assertTrue(walMetaData.getMemTablesId().isEmpty()); } + @Test + public void testInsertTabletEntrySerializedSizeWithRange() + throws IOException, IllegalPathException { + WALEntry walEntry = new WALInfoEntry(1, getInsertTabletNode(devicePath), 1, 3); + ByteBuffer byteBuffer = ByteBuffer.allocate(walEntry.serializedSize()); + WALByteBufferForTest buffer = new WALByteBufferForTest(byteBuffer); + walEntry.serialize(buffer); + assertEquals(walEntry.serializedSize(), byteBuffer.position()); + } + public static InsertRowNode getInsertRowNode(String devicePath) throws IllegalPathException { long time = 110L; TSDataType[] dataTypes = From dc0f370bb990deaafa76015cba67c7a19b7fd65e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 9 Jun 2026 18:29:00 +0800 Subject: [PATCH 2/4] Fix WAL charset deserialization --- .../plan/node/write/DeleteDataNode.java | 10 +- .../planner/plan/node/write/InsertNode.java | 5 +- .../plan/node/write/InsertRowNode.java | 7 +- .../plan/node/write/InsertTabletNode.java | 6 +- .../dataregion/memtable/AbstractMemTable.java | 5 +- .../memtable/WritableMemChunkGroup.java | 8 +- .../dataregion/modification/Deletion.java | 29 ++++- .../dataregion/wal/utils/WALReadUtils.java | 101 ++++++++++++++++++ 8 files changed, 147 insertions(+), 24 deletions(-) create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALReadUtils.java diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java index 3097a51fc50fc..dda2d245f0921 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java @@ -36,6 +36,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.tsfile.read.filter.factory.TimeFilterApi; @@ -153,10 +154,9 @@ public static DeleteDataNode deserializeFromWAL(DataInputStream stream) throws I for (int i = 0; i < size; i++) { try { pathList.add( - DataNodeDevicePathCache.getInstance() - .getPartialPath(ReadWriteIOUtils.readString(stream))); + DataNodeDevicePathCache.getInstance().getPartialPath(WALReadUtils.readString(stream))); } catch (IllegalPathException e) { - throw new IllegalArgumentException("Cannot deserialize InsertRowNode", e); + throw new IllegalArgumentException("Cannot deserialize DeleteDataNode", e); } } long deleteStartTime = stream.readLong(); @@ -174,9 +174,9 @@ public static DeleteDataNode deserializeFromWAL(ByteBuffer buffer) { List pathList = new ArrayList<>(size); for (int i = 0; i < size; i++) { try { - pathList.add(new PartialPath(ReadWriteIOUtils.readString(buffer))); + pathList.add(new PartialPath(WALReadUtils.readString(buffer))); } catch (IllegalPathException e) { - throw new IllegalArgumentException("Cannot deserialize InsertRowNode", e); + throw new IllegalArgumentException("Cannot deserialize DeleteDataNode", e); } } long deleteStartTime = buffer.getLong(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java index c664bbb9ddfaa..7d09a1c91aa4d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java @@ -31,6 +31,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.db.storageengine.dataregion.memtable.DeviceIDFactory; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.tsfile.enums.TSDataType; @@ -244,7 +245,7 @@ protected void serializeMeasurementSchemasToWAL(IWALByteBufferView buffer) { */ protected void deserializeMeasurementSchemas(DataInputStream stream) throws IOException { for (int i = 0; i < measurements.length; i++) { - measurementSchemas[i] = MeasurementSchema.deserializeFrom(stream); + measurementSchemas[i] = WALReadUtils.readMeasurementSchema(stream); measurements[i] = measurementSchemas[i].getMeasurementId(); dataTypes[i] = measurementSchemas[i].getType(); } @@ -252,7 +253,7 @@ protected void deserializeMeasurementSchemas(DataInputStream stream) throws IOEx protected void deserializeMeasurementSchemas(ByteBuffer buffer) { for (int i = 0; i < measurements.length; i++) { - measurementSchemas[i] = MeasurementSchema.deserializeFrom(buffer); + measurementSchemas[i] = WALReadUtils.readMeasurementSchema(buffer); measurements[i] = measurementSchemas[i].getMeasurementId(); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java index 41c4fbeca9be2..3790f9da4b97a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java @@ -34,6 +34,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.TypeInferenceUtils; @@ -671,8 +672,7 @@ protected static InsertRowNode subDeserializeFromWAL(DataInputStream stream) thr insertNode.setTime(stream.readLong()); try { insertNode.setDevicePath( - DataNodeDevicePathCache.getInstance() - .getPartialPath(ReadWriteIOUtils.readString(stream))); + DataNodeDevicePathCache.getInstance().getPartialPath(WALReadUtils.readString(stream))); } catch (IllegalPathException e) { throw new IllegalArgumentException(DESERIALIZE_ERROR, e); } @@ -757,8 +757,7 @@ protected static InsertRowNode subDeserializeFromWAL(ByteBuffer buffer) { insertNode.setTime(buffer.getLong()); try { insertNode.setDevicePath( - DataNodeDevicePathCache.getInstance() - .getPartialPath(ReadWriteIOUtils.readString(buffer))); + DataNodeDevicePathCache.getInstance().getPartialPath(WALReadUtils.readString(buffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException(DESERIALIZE_ERROR, e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index 283dba678e7e3..fcf1371b68a41 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java @@ -36,6 +36,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryValue; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.BitMapUtils; import org.apache.iotdb.db.utils.QueryDataSetUtils; @@ -943,7 +944,7 @@ private void subDeserializeFromWAL(DataInputStream stream) throws IOException { searchIndex = stream.readLong(); try { devicePath = - DataNodeDevicePathCache.getInstance().getPartialPath(ReadWriteIOUtils.readString(stream)); + DataNodeDevicePathCache.getInstance().getPartialPath(WALReadUtils.readString(stream)); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertTabletNode", e); } @@ -979,8 +980,7 @@ private void subDeserializeFromWAL(ByteBuffer buffer) { searchIndex = buffer.getLong(); try { devicePath = - DataNodeDevicePathCache.getInstance() - .getPartialPath((ReadWriteIOUtils.readString(buffer))); + DataNodeDevicePathCache.getInstance().getPartialPath((WALReadUtils.readString(buffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertTabletNode", e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index eb49d5325b0b3..ab95fc57662aa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -40,6 +40,7 @@ import org.apache.iotdb.db.storageengine.dataregion.read.filescan.impl.MemAlignedChunkHandleImpl; import org.apache.iotdb.db.storageengine.dataregion.read.filescan.impl.MemChunkHandleImpl; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.ModificationUtils; @@ -901,7 +902,7 @@ public int serializedSize() { } int size = FIXED_SERIALIZED_SIZE; for (Map.Entry entry : memTableMap.entrySet()) { - size += ReadWriteIOUtils.sizeToWrite(((PlainDeviceID) entry.getKey()).toStringID()); + size += WALWriteUtils.sizeToWrite(((PlainDeviceID) entry.getKey()).toStringID()); size += Byte.BYTES; size += entry.getValue().serializedSize(); } @@ -948,7 +949,7 @@ protected void deserialize(DataInputStream stream, boolean multiTvListMemChunk) int memTableMapSize = stream.readInt(); for (int i = 0; i < memTableMapSize; ++i) { - IDeviceID deviceID = deviceIDFactory.getDeviceID(ReadWriteIOUtils.readString(stream)); + IDeviceID deviceID = deviceIDFactory.getDeviceID(WALReadUtils.readString(stream)); boolean isAligned = ReadWriteIOUtils.readBool(stream); IWritableMemChunkGroup memChunkGroup; if (multiTvListMemChunk) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java index 226c2fd398098..ff519a2206aa1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/WritableMemChunkGroup.java @@ -22,10 +22,10 @@ import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.commons.path.PathPatternUtil; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.tsfile.utils.BitMap; -import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.schema.IMeasurementSchema; import java.io.DataInputStream; @@ -180,7 +180,7 @@ public int serializedSize() { int size = 0; size += Integer.BYTES; for (Map.Entry entry : memChunkMap.entrySet()) { - size += ReadWriteIOUtils.sizeToWrite(entry.getKey()); + size += WALWriteUtils.sizeToWrite(entry.getKey()); size += entry.getValue().serializedSize(); } return size; @@ -200,7 +200,7 @@ public static WritableMemChunkGroup deserialize(DataInputStream stream) throws I WritableMemChunkGroup memChunkGroup = new WritableMemChunkGroup(); int memChunkMapSize = stream.readInt(); for (int i = 0; i < memChunkMapSize; ++i) { - String measurement = ReadWriteIOUtils.readString(stream); + String measurement = WALReadUtils.readString(stream); IWritableMemChunk memChunk = WritableMemChunk.deserialize(stream); memChunkGroup.memChunkMap.put(measurement, memChunk); } @@ -212,7 +212,7 @@ public static WritableMemChunkGroup deserializeSingleTVListMemChunks(DataInputSt WritableMemChunkGroup memChunkGroup = new WritableMemChunkGroup(); int memChunkMapSize = stream.readInt(); for (int i = 0; i < memChunkMapSize; ++i) { - String measurement = ReadWriteIOUtils.readString(stream); + String measurement = WALReadUtils.readString(stream); IWritableMemChunk memChunk = WritableMemChunk.deserializeSingleTVListMemChunks(stream); memChunkGroup.memChunkMap.put(measurement, memChunk); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java index 1cb70a4c9cb3f..c4c74df1623e2 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/modification/Deletion.java @@ -28,7 +28,6 @@ import org.apache.tsfile.common.constant.TsFileConstant; import org.apache.tsfile.read.common.TimeRange; import org.apache.tsfile.utils.RamUsageEstimator; -import org.apache.tsfile.utils.ReadWriteIOUtils; import java.io.DataInputStream; import java.io.DataOutputStream; @@ -114,7 +113,7 @@ public long serializeWithoutFileOffset(DataOutputStream stream) throws IOExcepti serializeSize += Long.BYTES; stream.writeLong(getEndTime()); serializeSize += Long.BYTES; - serializeSize += ReadWriteIOUtils.write(getPathString(), stream); + serializeSize += writeString(getPathString(), stream); return serializeSize; } @@ -122,8 +121,30 @@ public static Deletion deserializeWithoutFileOffset(DataInputStream stream) throws IOException, IllegalPathException { long startTime = stream.readLong(); long endTime = stream.readLong(); - return new Deletion( - getMeasurementPath(ReadWriteIOUtils.readString(stream)), 0, startTime, endTime); + return new Deletion(getMeasurementPath(readString(stream)), 0, startTime, endTime); + } + + private static int writeString(String value, DataOutputStream stream) throws IOException { + if (value == null) { + stream.writeInt(-1); + return Integer.BYTES; + } + byte[] bytes = value.getBytes(TSFileConfig.STRING_CHARSET); + stream.writeInt(bytes.length); + stream.write(bytes); + return Integer.BYTES + bytes.length; + } + + private static String readString(DataInputStream stream) throws IOException { + int strLength = stream.readInt(); + if (strLength < 0) { + return null; + } else if (strLength == 0) { + return ""; + } + byte[] bytes = new byte[strLength]; + stream.readFully(bytes); + return new String(bytes, TSFileConfig.STRING_CHARSET); } private static PartialPath getMeasurementPath(String path) throws IllegalPathException { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALReadUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALReadUtils.java new file mode 100644 index 0000000000000..6d467f19934a5 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALReadUtils.java @@ -0,0 +1,101 @@ +/* + * 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.storageengine.dataregion.wal.utils; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.write.schema.MeasurementSchema; + +import java.io.DataInputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Map; + +/** Read methods paired with {@link WALWriteUtils}. */ +public class WALReadUtils { + + private WALReadUtils() {} + + public static String readString(DataInputStream stream) throws IOException { + int strLength = stream.readInt(); + if (strLength < 0) { + return null; + } else if (strLength == 0) { + return ""; + } + byte[] bytes = new byte[strLength]; + stream.readFully(bytes); + return new String(bytes, TSFileConfig.STRING_CHARSET); + } + + public static String readString(ByteBuffer buffer) { + int strLength = buffer.getInt(); + if (strLength < 0) { + return null; + } else if (strLength == 0) { + return ""; + } + byte[] bytes = new byte[strLength]; + buffer.get(bytes); + return new String(bytes, TSFileConfig.STRING_CHARSET); + } + + public static MeasurementSchema readMeasurementSchema(DataInputStream stream) throws IOException { + String measurementId = readString(stream); + byte type = stream.readByte(); + byte encoding = stream.readByte(); + byte compressor = stream.readByte(); + Map props = readProps(stream); + return new MeasurementSchema(measurementId, type, encoding, compressor, props); + } + + public static MeasurementSchema readMeasurementSchema(ByteBuffer buffer) { + String measurementId = readString(buffer); + byte type = buffer.get(); + byte encoding = buffer.get(); + byte compressor = buffer.get(); + Map props = readProps(buffer); + return new MeasurementSchema(measurementId, type, encoding, compressor, props); + } + + private static Map readProps(DataInputStream stream) throws IOException { + int size = stream.readInt(); + if (size <= 0) { + return null; + } + Map props = new HashMap<>(); + for (int i = 0; i < size; i++) { + props.put(readString(stream), readString(stream)); + } + return props; + } + + private static Map readProps(ByteBuffer buffer) { + int size = buffer.getInt(); + if (size <= 0) { + return null; + } + Map props = new HashMap<>(); + for (int i = 0; i < size; i++) { + props.put(readString(buffer), readString(buffer)); + } + return props; + } +} From 24afdd4fc38974b7e716b1a64f4994dce38421e0 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 10 Jun 2026 11:58:16 +0800 Subject: [PATCH 3/4] Fix insert node serde charset handling --- .../planner/plan/node/write/InsertNode.java | 100 ++++++++++++++++++ .../plan/node/write/InsertRowNode.java | 24 ++--- .../node/write/InsertRowsOfOneDeviceNode.java | 6 +- .../plan/node/write/InsertTabletNode.java | 18 ++-- 4 files changed, 124 insertions(+), 24 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java index 7d09a1c91aa4d..aee5cc1ec9384 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java @@ -34,9 +34,11 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALReadUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.NotImplementedException; import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.schema.MeasurementSchema; import java.io.DataInputStream; @@ -44,7 +46,9 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.stream.Collectors; @@ -214,6 +218,102 @@ protected void serializeAttributes(DataOutputStream stream) throws IOException { throw new NotImplementedException("serializeAttributes of InsertNode is not implemented"); } + protected static int serializeString(String value, ByteBuffer buffer) { + if (value == null) { + return ReadWriteIOUtils.write(-1, buffer); + } + byte[] bytes = value.getBytes(TSFileConfig.STRING_CHARSET); + int len = ReadWriteIOUtils.write(bytes.length, buffer); + buffer.put(bytes); + return len + bytes.length; + } + + protected static int serializeString(String value, DataOutputStream stream) throws IOException { + if (value == null) { + return ReadWriteIOUtils.write(-1, stream); + } + byte[] bytes = value.getBytes(TSFileConfig.STRING_CHARSET); + int len = ReadWriteIOUtils.write(bytes.length, stream); + stream.write(bytes); + return len + bytes.length; + } + + protected static String deserializeString(ByteBuffer buffer) { + int strLength = ReadWriteIOUtils.readInt(buffer); + if (strLength < 0) { + return null; + } else if (strLength == 0) { + return ""; + } + byte[] bytes = new byte[strLength]; + buffer.get(bytes); + return new String(bytes, TSFileConfig.STRING_CHARSET); + } + + protected static void serializeMeasurementSchema( + MeasurementSchema measurementSchema, ByteBuffer buffer) { + serializeString(measurementSchema.getMeasurementId(), buffer); + ReadWriteIOUtils.write(measurementSchema.getTypeInByte(), buffer); + ReadWriteIOUtils.write(measurementSchema.getEncodingType().serialize(), buffer); + ReadWriteIOUtils.write(measurementSchema.getCompressor().serialize(), buffer); + serializeProps(measurementSchema.getProps(), buffer); + } + + protected static void serializeMeasurementSchema( + MeasurementSchema measurementSchema, DataOutputStream stream) throws IOException { + serializeString(measurementSchema.getMeasurementId(), stream); + ReadWriteIOUtils.write(measurementSchema.getTypeInByte(), stream); + ReadWriteIOUtils.write(measurementSchema.getEncodingType().serialize(), stream); + ReadWriteIOUtils.write(measurementSchema.getCompressor().serialize(), stream); + serializeProps(measurementSchema.getProps(), stream); + } + + protected static MeasurementSchema deserializeMeasurementSchema(ByteBuffer buffer) { + String measurementId = deserializeString(buffer); + byte type = ReadWriteIOUtils.readByte(buffer); + byte encoding = ReadWriteIOUtils.readByte(buffer); + byte compressor = ReadWriteIOUtils.readByte(buffer); + Map props = deserializeProps(buffer); + return new MeasurementSchema(measurementId, type, encoding, compressor, props); + } + + private static void serializeProps(Map props, ByteBuffer buffer) { + if (props == null) { + ReadWriteIOUtils.write(0, buffer); + return; + } + ReadWriteIOUtils.write(props.size(), buffer); + for (Map.Entry entry : props.entrySet()) { + serializeString(entry.getKey(), buffer); + serializeString(entry.getValue(), buffer); + } + } + + private static void serializeProps(Map props, DataOutputStream stream) + throws IOException { + if (props == null) { + ReadWriteIOUtils.write(0, stream); + return; + } + ReadWriteIOUtils.write(props.size(), stream); + for (Map.Entry entry : props.entrySet()) { + serializeString(entry.getKey(), stream); + serializeString(entry.getValue(), stream); + } + } + + private static Map deserializeProps(ByteBuffer buffer) { + int size = ReadWriteIOUtils.readInt(buffer); + if (size <= 0) { + return null; + } + Map props = new HashMap<>(); + for (int i = 0; i < size; i++) { + props.put(deserializeString(buffer), deserializeString(buffer)); + } + return props; + } + // region Serialization methods for WAL /** Serialized size of measurement schemas, ignoring failed time series */ protected int serializeMeasurementSchemasSize() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java index 3790f9da4b97a..ba533ce5e4cae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java @@ -239,13 +239,13 @@ protected void serializeAttributes(DataOutputStream stream) throws IOException { void subSerialize(ByteBuffer buffer) { ReadWriteIOUtils.write(time, buffer); - ReadWriteIOUtils.write(devicePath.getFullPath(), buffer); + serializeString(devicePath.getFullPath(), buffer); serializeMeasurementsAndValues(buffer); } void subSerialize(DataOutputStream stream) throws IOException { ReadWriteIOUtils.write(time, stream); - ReadWriteIOUtils.write(devicePath.getFullPath(), stream); + serializeString(devicePath.getFullPath(), stream); serializeMeasurementsAndValues(stream); } @@ -282,9 +282,9 @@ private void serializeMeasurementsOrSchemas(ByteBuffer buffer) { } // serialize measurement schemas when exist if (measurementSchemas != null) { - measurementSchemas[i].serializeTo(buffer); + serializeMeasurementSchema(measurementSchemas[i], buffer); } else { - ReadWriteIOUtils.write(measurements[i], buffer); + serializeString(measurements[i], buffer); } } } @@ -304,9 +304,9 @@ private void serializeMeasurementsOrSchemas(DataOutputStream stream) throws IOEx } // serialize measurement schemas when exist if (measurementSchemas != null) { - measurementSchemas[i].serializeTo(stream); + serializeMeasurementSchema(measurementSchemas[i], stream); } else { - ReadWriteIOUtils.write(measurements[i], stream); + serializeString(measurements[i], stream); } } } @@ -332,7 +332,7 @@ private void putDataTypesAndValues(ByteBuffer buffer) { // and is forwarded to other nodes if (isNeedInferType) { ReadWriteIOUtils.write(TYPE_RAW_STRING, buffer); - ReadWriteIOUtils.write(values[i].toString(), buffer); + serializeString(values[i].toString(), buffer); } else { ReadWriteIOUtils.write(dataTypes[i], buffer); switch (dataTypes[i]) { @@ -387,7 +387,7 @@ private void putDataTypesAndValues(DataOutputStream stream) throws IOException { // and is forwarded to other nodes if (isNeedInferType) { ReadWriteIOUtils.write(TYPE_RAW_STRING, stream); - ReadWriteIOUtils.write(values[i].toString(), stream); + serializeString(values[i].toString(), stream); } else { ReadWriteIOUtils.write(dataTypes[i], stream); switch (dataTypes[i]) { @@ -433,7 +433,7 @@ void subDeserialize(ByteBuffer byteBuffer) { try { devicePath = DataNodeDevicePathCache.getInstance() - .getPartialPath(ReadWriteIOUtils.readString(byteBuffer)); + .getPartialPath(deserializeString(byteBuffer)); } catch (IllegalPathException e) { throw new IllegalArgumentException(DESERIALIZE_ERROR, e); } @@ -448,12 +448,12 @@ void deserializeMeasurementsAndValues(ByteBuffer buffer) { if (hasSchema) { measurementSchemas = new MeasurementSchema[measurementSize]; for (int i = 0; i < measurementSize; i++) { - measurementSchemas[i] = MeasurementSchema.deserializeFrom(buffer); + measurementSchemas[i] = deserializeMeasurementSchema(buffer); measurements[i] = measurementSchemas[i].getMeasurementId(); } } else { for (int i = 0; i < measurementSize; i++) { - measurements[i] = ReadWriteIOUtils.readString(buffer); + measurements[i] = deserializeString(buffer); } } @@ -477,7 +477,7 @@ private void fillDataTypesAndValues(ByteBuffer buffer) { // and is forwarded to other nodes byte typeNum = (byte) ReadWriteIOUtils.read(buffer); if (typeNum == TYPE_RAW_STRING || typeNum == TYPE_NULL) { - values[i] = typeNum == TYPE_RAW_STRING ? ReadWriteIOUtils.readString(buffer) : null; + values[i] = typeNum == TYPE_RAW_STRING ? deserializeString(buffer) : null; continue; } dataTypes[i] = TSDataType.values()[typeNum]; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java index c7b193841d45c..421524d4ae432 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java @@ -237,7 +237,7 @@ public static InsertRowsOfOneDeviceNode deserialize(ByteBuffer byteBuffer) { try { devicePath = DataNodeDevicePathCache.getInstance() - .getPartialPath((ReadWriteIOUtils.readString(byteBuffer))); + .getPartialPath((deserializeString(byteBuffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertRowsOfOneDeviceNode", e); } @@ -269,7 +269,7 @@ public static InsertRowsOfOneDeviceNode deserialize(ByteBuffer byteBuffer) { @Override protected void serializeAttributes(ByteBuffer byteBuffer) { PlanNodeType.INSERT_ROWS_OF_ONE_DEVICE.serialize(byteBuffer); - ReadWriteIOUtils.write(devicePath.getFullPath(), byteBuffer); + serializeString(devicePath.getFullPath(), byteBuffer); ReadWriteIOUtils.write(insertRowNodeList.size(), byteBuffer); @@ -285,7 +285,7 @@ protected void serializeAttributes(ByteBuffer byteBuffer) { @Override protected void serializeAttributes(DataOutputStream stream) throws IOException { PlanNodeType.INSERT_ROWS_OF_ONE_DEVICE.serialize(stream); - ReadWriteIOUtils.write(devicePath.getFullPath(), stream); + serializeString(devicePath.getFullPath(), stream); ReadWriteIOUtils.write(insertRowNodeList.size(), stream); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index fcf1371b68a41..c614d77e2cf47 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java @@ -414,7 +414,7 @@ protected void serializeAttributes(DataOutputStream stream) throws IOException { } void subSerialize(ByteBuffer buffer) { - ReadWriteIOUtils.write(devicePath.getFullPath(), buffer); + serializeString(devicePath.getFullPath(), buffer); writeMeasurementsOrSchemas(buffer); writeDataTypes(buffer); writeTimes(buffer); @@ -424,7 +424,7 @@ void subSerialize(ByteBuffer buffer) { } void subSerialize(DataOutputStream stream) throws IOException { - ReadWriteIOUtils.write(devicePath.getFullPath(), stream); + serializeString(devicePath.getFullPath(), stream); writeMeasurementsOrSchemas(stream); writeDataTypes(stream); writeTimes(stream); @@ -445,9 +445,9 @@ private void writeMeasurementsOrSchemas(ByteBuffer buffer) { } // serialize measurement schemas when exist if (measurementSchemas != null) { - measurementSchemas[i].serializeTo(buffer); + serializeMeasurementSchema(measurementSchemas[i], buffer); } else { - ReadWriteIOUtils.write(measurements[i], buffer); + serializeString(measurements[i], buffer); } } } @@ -464,9 +464,9 @@ private void writeMeasurementsOrSchemas(DataOutputStream stream) throws IOExcept } // serialize measurement schemas when exist if (measurementSchemas != null) { - measurementSchemas[i].serializeTo(stream); + serializeMeasurementSchema(measurementSchemas[i], stream); } else { - ReadWriteIOUtils.write(measurements[i], stream); + serializeString(measurements[i], stream); } } } @@ -684,7 +684,7 @@ public void subDeserialize(ByteBuffer buffer) { try { devicePath = DataNodeDevicePathCache.getInstance() - .getPartialPath((ReadWriteIOUtils.readString(buffer))); + .getPartialPath((deserializeString(buffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertTabletNode", e); } @@ -696,12 +696,12 @@ public void subDeserialize(ByteBuffer buffer) { if (hasSchema) { this.measurementSchemas = new MeasurementSchema[measurementSize]; for (int i = 0; i < measurementSize; i++) { - measurementSchemas[i] = MeasurementSchema.deserializeFrom(buffer); + measurementSchemas[i] = deserializeMeasurementSchema(buffer); measurements[i] = measurementSchemas[i].getMeasurementId(); } } else { for (int i = 0; i < measurementSize; i++) { - measurements[i] = ReadWriteIOUtils.readString(buffer); + measurements[i] = deserializeString(buffer); } } From 9e3ceb86c61cb12f7ec49639db2df667357be86c Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 10 Jun 2026 12:29:31 +0800 Subject: [PATCH 4/4] spotless --- .../plan/planner/plan/node/write/InsertRowNode.java | 3 +-- .../planner/plan/node/write/InsertRowsOfOneDeviceNode.java | 3 +-- .../plan/planner/plan/node/write/InsertTabletNode.java | 3 +-- 3 files changed, 3 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java index ba533ce5e4cae..23c17808d493c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java @@ -432,8 +432,7 @@ void subDeserialize(ByteBuffer byteBuffer) { time = byteBuffer.getLong(); try { devicePath = - DataNodeDevicePathCache.getInstance() - .getPartialPath(deserializeString(byteBuffer)); + DataNodeDevicePathCache.getInstance().getPartialPath(deserializeString(byteBuffer)); } catch (IllegalPathException e) { throw new IllegalArgumentException(DESERIALIZE_ERROR, e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java index 421524d4ae432..d7a6e8b6f5f59 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java @@ -236,8 +236,7 @@ public static InsertRowsOfOneDeviceNode deserialize(ByteBuffer byteBuffer) { try { devicePath = - DataNodeDevicePathCache.getInstance() - .getPartialPath((deserializeString(byteBuffer))); + DataNodeDevicePathCache.getInstance().getPartialPath((deserializeString(byteBuffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertRowsOfOneDeviceNode", e); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java index c614d77e2cf47..472c1e79d7e47 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java @@ -683,8 +683,7 @@ public static InsertTabletNode deserialize(ByteBuffer byteBuffer) { public void subDeserialize(ByteBuffer buffer) { try { devicePath = - DataNodeDevicePathCache.getInstance() - .getPartialPath((deserializeString(buffer))); + DataNodeDevicePathCache.getInstance().getPartialPath((deserializeString(buffer))); } catch (IllegalPathException e) { throw new IllegalArgumentException("Cannot deserialize InsertTabletNode", e); }