Skip to content

[to dev/1.3] Fix tree model WAL serialized sizes #17886

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 4 commits into from
Jun 10, 2026
Merged

[to dev/1.3] Fix tree model WAL serialized sizes #17886

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
09a5710
[to dev/1.3] Fix tree model WAL serialized sizes
Caideyipi Jun 9, 2026
dc0f370
Fix WAL charset deserialization
Caideyipi Jun 9, 2026
24afdd4
Fix insert node serde charset handling
Caideyipi Jun 10, 2026
9e3ceb8
spotless
Caideyipi Jun 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions ...ain/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/DeleteDataNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -129,7 +130,7 @@ public List<String> 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;
}
Expand All @@ -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();
Expand All @@ -174,9 +174,9 @@ public static DeleteDataNode deserializeFromWAL(ByteBuffer buffer) {
List<PartialPath> 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();
Expand Down
105 changes: 103 additions & 2 deletions ...rc/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,19 +31,24 @@
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.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;
import java.io.DataOutputStream;
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;

Expand Down Expand Up @@ -213,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<String, String> props = deserializeProps(buffer);
return new MeasurementSchema(measurementId, type, encoding, compressor, props);
}

private static void serializeProps(Map<String, String> props, ByteBuffer buffer) {
if (props == null) {
ReadWriteIOUtils.write(0, buffer);
return;
}
ReadWriteIOUtils.write(props.size(), buffer);
for (Map.Entry<String, String> entry : props.entrySet()) {
serializeString(entry.getKey(), buffer);
serializeString(entry.getValue(), buffer);
}
}

private static void serializeProps(Map<String, String> props, DataOutputStream stream)
throws IOException {
if (props == null) {
ReadWriteIOUtils.write(0, stream);
return;
}
ReadWriteIOUtils.write(props.size(), stream);
for (Map.Entry<String, String> entry : props.entrySet()) {
serializeString(entry.getKey(), stream);
serializeString(entry.getValue(), stream);
}
}

private static Map<String, String> deserializeProps(ByteBuffer buffer) {
int size = ReadWriteIOUtils.readInt(buffer);
if (size <= 0) {
return null;
}
Map<String, String> 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() {
Expand Down Expand Up @@ -244,15 +345,15 @@ 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();
}
}

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();
}
}
Expand Down
34 changes: 16 additions & 18 deletions ...main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -238,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);
}

Expand Down Expand Up @@ -281,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);
}
}
}
Expand All @@ -303,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);
}
}
}
Expand All @@ -331,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]) {
Expand Down Expand Up @@ -386,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]) {
Expand Down Expand Up @@ -431,8 +432,7 @@ void subDeserialize(ByteBuffer byteBuffer) {
time = byteBuffer.getLong();
try {
devicePath =
DataNodeDevicePathCache.getInstance()
.getPartialPath(ReadWriteIOUtils.readString(byteBuffer));
DataNodeDevicePathCache.getInstance().getPartialPath(deserializeString(byteBuffer));
} catch (IllegalPathException e) {
throw new IllegalArgumentException(DESERIALIZE_ERROR, e);
}
Expand All @@ -447,12 +447,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);
}
}

Expand All @@ -476,7 +476,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];
Expand Down Expand Up @@ -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();
}

Expand Down Expand Up @@ -671,8 +671,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);
}
Expand Down Expand Up @@ -757,8 +756,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);
}
Expand Down
7 changes: 3 additions & 4 deletions ...g/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsOfOneDeviceNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,8 +236,7 @@ public static InsertRowsOfOneDeviceNode deserialize(ByteBuffer byteBuffer) {

try {
devicePath =
DataNodeDevicePathCache.getInstance()
.getPartialPath((ReadWriteIOUtils.readString(byteBuffer)));
DataNodeDevicePathCache.getInstance().getPartialPath((deserializeString(byteBuffer)));
} catch (IllegalPathException e) {
throw new IllegalArgumentException("Cannot deserialize InsertRowsOfOneDeviceNode", e);
}
Expand Down Expand Up @@ -269,7 +268,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);

Expand All @@ -285,7 +284,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);

Expand Down
Loading
Loading