This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch cai/codex/session-batch-create-timeseries in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0b82bdfa2a2b11154a89950878eb3ce53908263d Author: Caideyipi <[email protected]> AuthorDate: Tue Oct 27 12:24:19 2026 +0800 Fix aligned batch schema plan handling --- .../write/InternalCreateMultiTimeSeriesNode.java | 52 +++++++++++++++- .../plan/node/metadata/write/MeasurementGroup.java | 71 +++++++++++++++++++--- .../metadata/write/MetadataWriteNodeSerDeTest.java | 2 + 3 files changed, 115 insertions(+), 10 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/InternalCreateMultiTimeSeriesNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/InternalCreateMultiTimeSeriesNode.java index 8e01176c415..f59df548b15 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/InternalCreateMultiTimeSeriesNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/InternalCreateMultiTimeSeriesNode.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.IPlanVisitor; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; +import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode; @@ -44,6 +45,9 @@ import java.util.Objects; public class InternalCreateMultiTimeSeriesNode extends WritePlanNode { + private static final int SPLIT_SIZE = + IoTDBDescriptor.getInstance().getConfig().getMaxMeasurementNumOfInternalRequest(); + // <DevicePath, <IsAligned, MeasurementGroup>> private Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap; @@ -161,10 +165,54 @@ public class InternalCreateMultiTimeSeriesNode extends WritePlanNode { List<WritePlanNode> result = new ArrayList<>(); for (Map.Entry<TRegionReplicaSet, Map<PartialPath, Pair<Boolean, MeasurementGroup>>> entry : splitMap.entrySet()) { - result.add( - new InternalCreateMultiTimeSeriesNode(getPlanNodeId(), entry.getValue(), entry.getKey())); + for (Map<PartialPath, Pair<Boolean, MeasurementGroup>> measurementGroupMap : + splitAndRegroupMeasurements(entry.getValue())) { + result.add( + new InternalCreateMultiTimeSeriesNode( + getPlanNodeId(), measurementGroupMap, entry.getKey())); + } + } + + return result; + } + + private static List<Map<PartialPath, Pair<Boolean, MeasurementGroup>>> + splitAndRegroupMeasurements(Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap) { + List<Map<PartialPath, Pair<Boolean, MeasurementGroup>>> result = new ArrayList<>(); + int measurementCount = 0; + Map<PartialPath, Pair<Boolean, MeasurementGroup>> currentMap = new HashMap<>(); + for (Map.Entry<PartialPath, Pair<Boolean, MeasurementGroup>> entry : deviceMap.entrySet()) { + final Pair<Boolean, MeasurementGroup> device = entry.getValue(); + if (device.right.size() > SPLIT_SIZE) { + for (MeasurementGroup splitMeasurementGroup : device.right.split(SPLIT_SIZE)) { + if (splitMeasurementGroup.size() == SPLIT_SIZE) { + Map<PartialPath, Pair<Boolean, MeasurementGroup>> singletonMap = new HashMap<>(); + singletonMap.put(entry.getKey(), new Pair<>(device.left, splitMeasurementGroup)); + result.add(singletonMap); + } else { + if (measurementCount + splitMeasurementGroup.size() > SPLIT_SIZE) { + result.add(currentMap); + currentMap = new HashMap<>(); + measurementCount = 0; + } + measurementCount += splitMeasurementGroup.size(); + currentMap.put(entry.getKey(), new Pair<>(device.left, splitMeasurementGroup)); + } + } + } else { + if (measurementCount + device.right.size() > SPLIT_SIZE) { + result.add(currentMap); + currentMap = new HashMap<>(); + measurementCount = 0; + } + measurementCount += device.right.size(); + currentMap.put(entry.getKey(), device); + } } + if (!currentMap.isEmpty()) { + result.add(currentMap); + } return result; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/MeasurementGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/MeasurementGroup.java index 699b904b669..3fa06cdd7e4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/MeasurementGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/metadata/write/MeasurementGroup.java @@ -36,6 +36,8 @@ import java.util.Set; public class MeasurementGroup { + private static final int PROPS_LIST_SERIALIZATION_MARKER = -1; + private List<String> measurements = new ArrayList<>(); private List<TSDataType> dataTypes = new ArrayList<>(); private List<TSEncoding> encodings = new ArrayList<>(); @@ -202,22 +204,23 @@ public class MeasurementGroup { private MeasurementGroup getSubMeasurementGroup(int startIndex, int endIndex) { MeasurementGroup subMeasurementGroup; subMeasurementGroup = new MeasurementGroup(); - subMeasurementGroup.measurements = measurements.subList(startIndex, endIndex); + subMeasurementGroup.measurements = new ArrayList<>(measurements.subList(startIndex, endIndex)); subMeasurementGroup.measurementSet.addAll(subMeasurementGroup.measurements); - subMeasurementGroup.dataTypes = dataTypes.subList(startIndex, endIndex); - subMeasurementGroup.encodings = encodings.subList(startIndex, endIndex); - subMeasurementGroup.compressors = compressors.subList(startIndex, endIndex); + subMeasurementGroup.dataTypes = new ArrayList<>(dataTypes.subList(startIndex, endIndex)); + subMeasurementGroup.encodings = new ArrayList<>(encodings.subList(startIndex, endIndex)); + subMeasurementGroup.compressors = new ArrayList<>(compressors.subList(startIndex, endIndex)); if (aliasList != null) { - subMeasurementGroup.aliasList = aliasList.subList(startIndex, endIndex); + subMeasurementGroup.aliasList = new ArrayList<>(aliasList.subList(startIndex, endIndex)); } if (propsList != null) { - subMeasurementGroup.propsList = propsList.subList(startIndex, endIndex); + subMeasurementGroup.propsList = new ArrayList<>(propsList.subList(startIndex, endIndex)); } if (tagsList != null) { - subMeasurementGroup.tagsList = tagsList.subList(startIndex, endIndex); + subMeasurementGroup.tagsList = new ArrayList<>(tagsList.subList(startIndex, endIndex)); } if (attributesList != null) { - subMeasurementGroup.attributesList = attributesList.subList(startIndex, endIndex); + subMeasurementGroup.attributesList = + new ArrayList<>(attributesList.subList(startIndex, endIndex)); } return subMeasurementGroup; } @@ -279,6 +282,11 @@ public class MeasurementGroup { ReadWriteIOUtils.write(attributes, byteBuffer); } } + + // Props were added after the original MeasurementGroup wire format. Keep the marker so old + // consensus requests can still be deserialized without treating the following field as props. + ReadWriteIOUtils.write(PROPS_LIST_SERIALIZATION_MARKER, byteBuffer); + serializePropsList(byteBuffer); } public void serialize(DataOutputStream stream) throws IOException { @@ -338,6 +346,35 @@ public class MeasurementGroup { ReadWriteIOUtils.write(attributes, stream); } } + + ReadWriteIOUtils.write(PROPS_LIST_SERIALIZATION_MARKER, stream); + serializePropsList(stream); + } + + private void serializePropsList(ByteBuffer byteBuffer) { + if (propsList == null) { + byteBuffer.put((byte) -1); + } else if (propsList.isEmpty()) { + byteBuffer.put((byte) 0); + } else { + byteBuffer.put((byte) 1); + for (Map<String, String> props : propsList) { + ReadWriteIOUtils.write(props, byteBuffer); + } + } + } + + private void serializePropsList(DataOutputStream stream) throws IOException { + if (propsList == null) { + stream.write((byte) -1); + } else if (propsList.isEmpty()) { + stream.write((byte) 0); + } else { + stream.write((byte) 1); + for (Map<String, String> props : propsList) { + ReadWriteIOUtils.write(props, stream); + } + } } public void deserialize(ByteBuffer byteBuffer) { @@ -392,6 +429,24 @@ public class MeasurementGroup { } } } + + // The next field is the plan node's data in the original format. Rewind when reading an old + // request so adding props does not make existing consensus logs unreadable. + final int propsMarkerPosition = byteBuffer.position(); + if (byteBuffer.remaining() >= Integer.BYTES + && ReadWriteIOUtils.readInt(byteBuffer) == PROPS_LIST_SERIALIZATION_MARKER) { + label = byteBuffer.get(); + if (label >= 0) { + propsList = new ArrayList<>(label == 1 ? size : 0); + if (label == 1) { + for (int i = 0; i < size; i++) { + propsList.add(ReadWriteIOUtils.readMap(byteBuffer)); + } + } + } + } else { + byteBuffer.position(propsMarkerPosition); + } } // This won't be affected by "removeMeasurements" diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/metadata/write/MetadataWriteNodeSerDeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/metadata/write/MetadataWriteNodeSerDeTest.java index 3ac85f40b42..4e7cda09865 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/metadata/write/MetadataWriteNodeSerDeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/metadata/write/MetadataWriteNodeSerDeTest.java @@ -60,6 +60,7 @@ import org.junit.Test; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -236,6 +237,7 @@ public class MetadataWriteNodeSerDeTest { PlanNodeId planNodeId = new PlanNodeId("InternalCreateMultiTimeSeriesNode"); MeasurementGroup measurementGroup = new MeasurementGroup(); measurementGroup.addMeasurement("s1", TSDataType.INT64, TSEncoding.PLAIN, CompressionType.GZIP); + measurementGroup.addProps(Collections.singletonMap("max_point_number", "3")); Map<PartialPath, Pair<Boolean, MeasurementGroup>> deviceMap = new HashMap<>(); deviceMap.put(new PartialPath("root.db"), new Pair<>(false, measurementGroup)); InternalCreateMultiTimeSeriesNode internalCreateMultiTimeSeriesNode =
