This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch table_insert_tablets in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 3a15f13fddce5c852865421c078e9d5f0723a5b2 Author: Tian Jiang <[email protected]> AuthorDate: Thu Jul 9 15:22:11 2026 +0800 Add RelationalInsertMultiTabletsNode --- .../tablet/PipeInsertNodeTabletInsertionEvent.java | 9 ++ .../resource/memory/InsertNodeMemoryEstimator.java | 20 +++ .../plan/node/DataNodePlanNodeDeserializer.java | 3 + .../plan/planner/plan/node/PlanVisitor.java | 5 + .../write/RelationalInsertMultiTabletsNode.java | 137 +++++++++++++++++++++ .../node/write/RelationalInsertTabletNode.java | 27 +++- .../plan/relational/planner/RelationPlanner.java | 6 +- .../consensus/ConsensusLogToTabletConverter.java | 1 + .../write/InsertMultiTabletsNodeSerdeTest.java | 56 +++++++++ .../plan/planner/plan/node/PlanNodeType.java | 1 + 10 files changed, 259 insertions(+), 6 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java index 5fab6979e29..f89cae3576e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeInsertNodeTabletInsertionEvent.java @@ -50,6 +50,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -549,6 +550,14 @@ public class PipeInsertNodeTabletInsertionEvent extends PipeInsertionEvent pipeTaskMeta, this, insertRowNode, tablePattern)); } break; + case RELATIONAL_INSERT_MULTI_TABLETS: + for (final InsertTabletNode insertTabletNode : + ((RelationalInsertMultiTabletsNode) node).getInsertTabletNodeList()) { + eventParsers.add( + new TabletInsertionEventTablePatternParser( + pipeTaskMeta, this, insertTabletNode, tablePattern)); + } + break; default: throw new UnSupportedDataTypeException( DataNodePipeMessages.UNSUPPORTED_NODE_TYPE + node.getType()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java index 54655704c1c..be1db2e3ba1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimator.java @@ -35,6 +35,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNod import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -71,6 +72,8 @@ public class InsertNodeMemoryEstimator { private static final String RELATIONAL_INSERT_ROWS_NODE = "RelationalInsertRowsNode"; private static final String RELATIONAL_INSERT_ROW_NODE = "RelationalInsertRowNode"; private static final String RELATIONAL_INSERT_TABLET_NODE = "RelationalInsertTabletNode"; + private static final String RELATIONAL_INSERT_MULTI_TABLETS_NODE = + "RelationalInsertMultiTabletsNode"; private static final long NUM_BYTES_OBJECT_REF = RamUsageEstimator.NUM_BYTES_OBJECT_REF; private static final long NUM_BYTES_OBJECT_HEADER = RamUsageEstimator.NUM_BYTES_OBJECT_HEADER; @@ -105,6 +108,9 @@ public class InsertNodeMemoryEstimator { private static final long RELATIONAL_INSERT_TABLET_NODE_SIZE = RamUsageEstimator.shallowSizeOfInstance(RelationalInsertTabletNode.class); + private static final long RELATIONAL_INSERT_MULTI_TABLETS_NODE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(RelationalInsertMultiTabletsNode.class); + // ============================Device And Measurement=================================== private static final long PARTIAL_PATH_SIZE = @@ -170,6 +176,9 @@ public class InsertNodeMemoryEstimator { return sizeOfInsertRowsOfOneDeviceNode((InsertRowsOfOneDeviceNode) insertNode); case INSERT_MULTI_TABLETS_NODE: return sizeOfInsertMultiTabletsNode((InsertMultiTabletsNode) insertNode); + case RELATIONAL_INSERT_MULTI_TABLETS_NODE: + return sizeOfRelationalInsertMultiTabletsNode( + (RelationalInsertMultiTabletsNode) insertNode); case RELATIONAL_INSERT_ROWS_NODE: return sizeOfRelationalInsertRowsNode((RelationalInsertRowsNode) insertNode); case RELATIONAL_INSERT_ROW_NODE: @@ -286,6 +295,17 @@ public class InsertNodeMemoryEstimator { return size; } + private static long sizeOfRelationalInsertMultiTabletsNode( + final RelationalInsertMultiTabletsNode node) { + final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet(); + long size = RELATIONAL_INSERT_MULTI_TABLETS_NODE_SIZE; + size += calculateFullInsertNodeSize(node, deduplicatedObjects); + size += sizeOfInsertTabletNodeList(node.getInsertTabletNodeList(), deduplicatedObjects); + size += sizeOfIntegerList(node.getParentInsertTabletNodeIndexList()); + size += sizeOfResults(node.getResults()); + return size; + } + private static long sizeOfRelationalInsertRowNode(final RelationalInsertRowNode node) { return sizeOfRelationalInsertRowNode(node, newDeduplicatedObjectSet()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/DataNodePlanNodeDeserializer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/DataNodePlanNodeDeserializer.java index 3e5e46a3c0c..e7e522e370a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/DataNodePlanNodeDeserializer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/DataNodePlanNodeDeserializer.java @@ -119,6 +119,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOf import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -481,6 +482,8 @@ public class DataNodePlanNodeDeserializer extends CommonPlanNodeDeserializer { return RelationalDeleteDataNode.deserialize(buffer); case 2004: return ObjectNode.deserialize(buffer); + case 2005: + return RelationalInsertMultiTabletsNode.deserialize(buffer); default: return super.deserialize(buffer, nodeType); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java index 81596f2d38c..5a17a68e0af 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java @@ -117,6 +117,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOf import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -605,6 +606,10 @@ public interface PlanVisitor<R, C> extends ICoreQueryPlanVisitor<R, C> { return visitPlan(node, context); } + default R visitRelationalInsertMultiTablets(RelationalInsertMultiTabletsNode node, C context) { + return visitInsertMultiTablets(node, context); + } + default R visitInsertRowsOfOneDevice(InsertRowsOfOneDeviceNode node, C context) { return visitPlan(node, context); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertMultiTabletsNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertMultiTabletsNode.java new file mode 100644 index 00000000000..7ce55c4cc87 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertMultiTabletsNode.java @@ -0,0 +1,137 @@ +/* + * 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.queryengine.plan.planner.plan.node.write; + +import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.IPlanVisitor; +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.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; + +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.DataOutputStream; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class RelationalInsertMultiTabletsNode extends InsertMultiTabletsNode { + + public RelationalInsertMultiTabletsNode(PlanNodeId id) { + super(id); + } + + public RelationalInsertMultiTabletsNode( + PlanNodeId id, + List<Integer> parentInsertTabletNodeIndexList, + List<InsertTabletNode> insertTabletNodeList) { + super(id, parentInsertTabletNodeIndexList, insertTabletNodeList); + } + + @Override + public PlanNodeType getType() { + return PlanNodeType.RELATIONAL_INSERT_MULTI_TABLETS; + } + + @Override + public List<WritePlanNode> splitByPartition(IAnalysis analysis) { + Map<TRegionReplicaSet, RelationalInsertMultiTabletsNode> splitMap = new HashMap<>(); + for (int i = 0; i < insertTabletNodeList.size(); i++) { + InsertTabletNode insertTabletNode = insertTabletNodeList.get(i); + List<WritePlanNode> tmpResult = insertTabletNode.splitByPartition(analysis); + for (WritePlanNode subNode : tmpResult) { + TRegionReplicaSet dataRegionReplicaSet = ((InsertNode) subNode).getDataRegionReplicaSet(); + RelationalInsertMultiTabletsNode tmpNode = splitMap.get(dataRegionReplicaSet); + if (tmpNode == null) { + tmpNode = new RelationalInsertMultiTabletsNode(this.getPlanNodeId()); + tmpNode.setDataRegionReplicaSet(dataRegionReplicaSet); + tmpNode.setPhysicalTime(getPhysicalTime()); + tmpNode.setNodeId(getNodeId()); + tmpNode.setSyncIndex(getSyncIndex()); + splitMap.put(dataRegionReplicaSet, tmpNode); + } + tmpNode.addInsertTabletNode((InsertTabletNode) subNode, i); + } + } + return new ArrayList<>(splitMap.values()); + } + + public static RelationalInsertMultiTabletsNode deserialize(ByteBuffer byteBuffer) { + PlanNodeId planNodeId; + List<InsertTabletNode> insertTabletNodeList = new ArrayList<>(); + List<Integer> parentIndex = new ArrayList<>(); + + int size = byteBuffer.getInt(); + for (int i = 0; i < size; i++) { + RelationalInsertTabletNode insertTabletNode = + new RelationalInsertTabletNode(new PlanNodeId("")); + insertTabletNode.subDeserialize(byteBuffer); + insertTabletNodeList.add(insertTabletNode); + } + for (int i = 0; i < size; i++) { + parentIndex.add(byteBuffer.getInt()); + } + + planNodeId = PlanNodeId.deserialize(byteBuffer); + for (InsertTabletNode insertTabletNode : insertTabletNodeList) { + insertTabletNode.setPlanNodeId(planNodeId); + } + + return new RelationalInsertMultiTabletsNode(planNodeId, parentIndex, insertTabletNodeList); + } + + @Override + protected void serializeAttributes(ByteBuffer byteBuffer) { + PlanNodeType.RELATIONAL_INSERT_MULTI_TABLETS.serialize(byteBuffer); + + ReadWriteIOUtils.write(insertTabletNodeList.size(), byteBuffer); + + for (InsertTabletNode node : insertTabletNodeList) { + node.subSerialize(byteBuffer); + } + for (Integer index : parentInsertTabletNodeIndexList) { + ReadWriteIOUtils.write(index, byteBuffer); + } + } + + @Override + protected void serializeAttributes(DataOutputStream stream) throws IOException { + PlanNodeType.RELATIONAL_INSERT_MULTI_TABLETS.serialize(stream); + + ReadWriteIOUtils.write(insertTabletNodeList.size(), stream); + + for (InsertTabletNode node : insertTabletNodeList) { + node.subSerialize(stream); + } + for (Integer index : parentInsertTabletNodeIndexList) { + ReadWriteIOUtils.write(index, stream); + } + } + + @Override + public <R, C> R accept(IPlanVisitor<R, C> visitor, C context) { + return ((PlanVisitor<R, C>) visitor).visitRelationalInsertMultiTablets(this, context); + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java index d41b078cb9b..f736e1a17c4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java @@ -96,6 +96,17 @@ public class RelationalInsertTabletNode extends InsertTabletNode { super(id); } + @Override + public InsertNode mergeInsertNode(List<InsertNode> insertNodes) { + List<Integer> index = new ArrayList<>(); + List<InsertTabletNode> insertTabletNodes = new ArrayList<>(); + for (int i = 0; i < insertNodes.size(); i++) { + insertTabletNodes.add((InsertTabletNode) insertNodes.get(i)); + index.add(i); + } + return new RelationalInsertMultiTabletsNode(this.getPlanNodeId(), index, insertTabletNodes); + } + @TestOnly public RelationalInsertTabletNode( PlanNodeId id, @@ -307,16 +318,26 @@ public class RelationalInsertTabletNode extends InsertTabletNode { @Override protected void serializeAttributes(ByteBuffer byteBuffer) { super.serializeAttributes(byteBuffer); + } + + @Override + protected void serializeAttributes(DataOutputStream stream) throws IOException { + super.serializeAttributes(stream); + } + + @Override + void subSerialize(ByteBuffer buffer) { + super.subSerialize(buffer); for (int i = 0; measurements != null && i < measurements.length; i++) { if (shouldSerializeMeasurement(i)) { - columnCategories[i].serialize(byteBuffer); + columnCategories[i].serialize(buffer); } } } @Override - protected void serializeAttributes(DataOutputStream stream) throws IOException { - super.serializeAttributes(stream); + void subSerialize(DataOutputStream stream) throws IOException { + super.subSerialize(stream); for (int i = 0; measurements != null && i < measurements.length; i++) { if (shouldSerializeMeasurement(i)) { columnCategories[i].serialize(stream); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java index d77f4f3f118..97e108c026c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java @@ -97,10 +97,10 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFileNod import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedDeleteDataNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedInsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.pipe.PipeEnrichedWritePlanNode; -import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; @@ -1381,8 +1381,8 @@ public class RelationPlanner implements AstVisitor<RelationPlan, Void> { @Override public RelationPlan visitInsertTablets(InsertTablets node, Void context) { final InsertMultiTabletsStatement insertMultiTabletsStatement = node.getInnerTreeStatement(); - final InsertMultiTabletsNode insertMultiTabletsNode = - new InsertMultiTabletsNode(idAllocator.genPlanNodeId()); + final RelationalInsertMultiTabletsNode insertMultiTabletsNode = + new RelationalInsertMultiTabletsNode(idAllocator.genPlanNodeId()); for (int i = 0; i < insertMultiTabletsStatement.getInsertTabletStatementList().size(); i++) { insertMultiTabletsNode.addInsertTabletNode( fromInsertTabletStatement( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java index df133274075..a933e86c74d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java @@ -205,6 +205,7 @@ public class ConsensusLogToTabletConverter { case INSERT_ROWS_OF_ONE_DEVICE: return convertInsertRowsOfOneDeviceNode((InsertRowsOfOneDeviceNode) insertNode); case INSERT_MULTI_TABLET: + case RELATIONAL_INSERT_MULTI_TABLETS: return convertInsertMultiTabletsNode((InsertMultiTabletsNode) insertNode); case RELATIONAL_INSERT_ROW: return convertRelationalInsertRowNode((RelationalInsertRowNode) insertNode); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertMultiTabletsNodeSerdeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertMultiTabletsNodeSerdeTest.java index b71c6dea18a..52d8b08bb91 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertMultiTabletsNodeSerdeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/write/InsertMultiTabletsNodeSerdeTest.java @@ -23,8 +23,11 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.path.PartialPath; 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.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertMultiTabletsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.enums.TSDataType; @@ -93,4 +96,57 @@ public class InsertMultiTabletsNodeSerdeTest { Assert.assertEquals(insertMultiTabletsNode, InsertMultiTabletsNode.deserialize(byteBuffer)); } + + @Test + public void testRelationalInsertMultiTabletPlan() { + long[] times = new long[] {110L, 111L, 112L, 113L}; + TSDataType[] dataTypes = + new TSDataType[] {TSDataType.STRING, TSDataType.STRING, TSDataType.DOUBLE}; + TsTableColumnCategory[] columnCategories = + new TsTableColumnCategory[] { + TsTableColumnCategory.TAG, TsTableColumnCategory.ATTRIBUTE, TsTableColumnCategory.FIELD + }; + + Object[] columns = new Object[3]; + columns[0] = new Binary[4]; + columns[1] = new Binary[4]; + columns[2] = new double[4]; + + for (int r = 0; r < 4; r++) { + ((Binary[]) columns[0])[r] = new Binary("tag" + r, TSFileConfig.STRING_CHARSET); + ((Binary[]) columns[1])[r] = new Binary("attr" + r, TSFileConfig.STRING_CHARSET); + ((double[]) columns[2])[r] = r; + } + + PlanNodeId planNodeId = new PlanNodeId("plan node 2"); + + RelationalInsertMultiTabletsNode insertMultiTabletsNode = + new RelationalInsertMultiTabletsNode(planNodeId); + + for (int i = 0; i < 10; i++) { + insertMultiTabletsNode.addInsertTabletNode( + new RelationalInsertTabletNode( + planNodeId, + new PartialPath("table" + i, false), + false, + new String[] {"tag1", "attr1", "s1"}, + dataTypes, + times, + null, + columns, + times.length, + columnCategories), + i); + } + + ByteBuffer byteBuffer = ByteBuffer.allocate(10000); + insertMultiTabletsNode.serialize(byteBuffer); + byteBuffer.flip(); + + Assert.assertEquals( + PlanNodeType.RELATIONAL_INSERT_MULTI_TABLETS.getNodeType(), byteBuffer.getShort()); + + Assert.assertEquals( + insertMultiTabletsNode, RelationalInsertMultiTabletsNode.deserialize(byteBuffer)); + } } diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java index e36796fca43..0fd5ff14913 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/queryengine/plan/planner/plan/node/PlanNodeType.java @@ -209,6 +209,7 @@ public enum PlanNodeType { RELATIONAL_INSERT_ROWS((short) 2002), RELATIONAL_DELETE_DATA((short) 2003), OBJECT_FILE_NODE((short) 2004), + RELATIONAL_INSERT_MULTI_TABLETS((short) 2005), ; private static final IPlanNodeDeserializer DESERIALIZER;
