You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/07/09 07:22:33 UTC
(iotdb) branch table_insert_tablets created (now 3a15f13fddc)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a change to branch table_insert_tablets
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 3a15f13fddc Add RelationalInsertMultiTabletsNode
This branch includes the following new commits:
new afbea6c558d Merge branch 'master' into table_insert_tablets
new 3a15f13fddc Add RelationalInsertMultiTabletsNode
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iotdb) 01/02: Merge branch 'master' into table_insert_tablets
Posted by ji...@apache.org.
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 afbea6c558d18022f8d48289776a8e4e51e54dc9
Merge: e7eefcdc1c5 f3e161218d6
Author: Tian Jiang <jt...@163.com>
AuthorDate: Thu Jul 9 11:35:29 2026 +0800
Merge branch 'master' into table_insert_tablets
.../it/database/IoTDBDatabaseRegionControlIT.java | 111 ++++
.../commit/IoTDBMigrateMultiRegionForIoTV1IT.java | 66 +++
...TDBRegionMigrateWithDeletionMultiDataDirIT.java | 14 +-
...gionMigrateWithDeletionMultiDataDirTableIT.java | 14 +-
.../RegionMigrateFileAssertions.java | 2 +-
.../IoTDBRegionMigrateNormalITForIoTV2BatchIT.java | 34 ++
.../manual/enhanced/IoTDBPipeDoubleLivingIT.java | 10 +-
.../IoTDBPipeReceiverAutoCreateDisabledIT.java | 37 ++
.../relational/it/db/it/IoTDBDeletionTableIT.java | 565 ++++++++++++++++++++-
.../it/db/it/IoTDBLoadConfigurationTableIT.java | 110 ++++
.../schema/IoTDBDatabaseMaxRegionGroupNumIT.java | 73 +++
.../java/org/apache/iotdb/jdbc/StringUtils.java | 21 +-
.../org/apache/iotdb/jdbc/StringUtilsTest.java | 74 +++
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 2 +
.../load/cache/region/RegionGroupCache.java | 8 +-
.../pipe/coordinator/task/PipeTaskCoordinator.java | 30 +-
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 56 +-
.../pipe/source/IoTDBConfigRegionSource.java | 21 +-
.../manager/schema/ClusterSchemaManager.java | 36 +-
.../procedure/impl/StateMachineProcedure.java | 13 +-
.../PipeConfigTreePrivilegeParseVisitorTest.java | 43 ++
.../confignode/procedure/STMProcedureTest.java | 70 +++
.../pipe/AbstractOperatePipeProcedureV2Test.java | 3 +-
.../consensus/iot/IoTConsensusServerImpl.java | 13 +-
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 4 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 18 +-
.../iotdb/db/i18n/DataNodeQueryMessages.java | 70 ++-
.../iotdb/db/i18n/StorageEngineMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 2 +-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 18 +-
.../iotdb/db/i18n/DataNodeQueryMessages.java | 58 ++-
.../iotdb/db/i18n/StorageEngineMessages.java | 2 +
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 38 +-
.../DirectBufferMemoryAllocationException.java} | 24 +-
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 455 +++--------------
.../task/builder/PipeDataNodeTaskBuilder.java | 36 ++
.../agent/task/connection/PipeEventCollector.java | 15 +-
.../agent/task/stage/PipeTaskProcessorStage.java | 5 +-
.../subtask/processor/PipeProcessorSubtask.java | 2 +
.../common/tsfile/PipeTsFileInsertionEvent.java | 144 ++++--
.../tsfile/parser/TsFileInsertionEventParser.java | 12 +-
.../query/TsFileInsertionEventQueryParser.java | 1 +
.../scan/TsFileInsertionEventScanParser.java | 14 +-
.../table/TsFileInsertionEventTableParser.java | 22 +-
.../processor/aggregate/AggregateProcessor.java | 3 +
.../downsampling/DownSamplingProcessor.java | 3 +
.../receiver/TwoStageAggregateReceiver.java | 10 +
.../exchange/sender/TwoStageAggregateSender.java | 75 ++-
.../twostage/plugin/TwoStageCountProcessor.java | 9 +-
.../protocol/thrift/IoTDBDataNodeReceiver.java | 178 ++++++-
.../db/pipe/resource/memory/PipeMemoryManager.java | 95 +++-
.../evolvable/batch/PipeTabletEventBatch.java | 69 ++-
.../evolvable/batch/PipeTabletEventPlainBatch.java | 61 ++-
.../batch/PipeTabletEventTsFileBatch.java | 30 +-
.../protocol/airgap/IoTDBDataRegionAirGapSink.java | 18 +-
.../iotconsensusv2/IoTConsensusV2SyncSink.java | 20 +-
.../IoTConsensusV2TsFileInsertionEventHandler.java | 34 +-
.../IoTConsensusV2TransferBatchReqBuilder.java | 109 ++--
.../sink/protocol/legacy/IoTDBLegacyPipeSink.java | 17 +-
.../async/handler/PipeTransferTsFileHandler.java | 42 +-
.../thrift/sync/IoTDBDataRegionSyncSink.java | 136 +++++
...istoricalDataRegionTsFileAndDeletionSource.java | 31 +-
.../realtime/PipeRealtimeDataRegionSource.java | 17 +-
.../schemaregion/IoTDBSchemaRegionSource.java | 4 +-
.../PipePlanTablePrivilegeParseVisitor.java | 17 +
.../protocol/thrift/impl/ClientRPCServiceImpl.java | 42 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 4 +-
.../db/queryengine/common/MPPQueryContext.java | 286 ++++++++++-
.../common/schematree/ClusterSchemaTree.java | 8 +
.../db/queryengine/plan/analyze/AnalyzeUtils.java | 436 ++++++++++++++--
.../queryengine/plan/analyze/AnalyzeVisitor.java | 8 +
.../queryengine/plan/analyze/ExpressionUtils.java | 2 +
.../analyze/schema/ClusterSchemaFetchExecutor.java | 9 +
.../config/executor/ClusterConfigTaskExecutor.java | 25 +-
.../relational/analyzer/StatementAnalyzer.java | 2 +-
.../read_tsfile/ExternalTsFileQueryResource.java | 9 +-
.../tvf/read_tsfile/ReadTsFileTableFunction.java | 60 ++-
.../plan/relational/planner/RelationPlanner.java | 5 +-
.../plan/relational/sql/ast/DeleteDevice.java | 5 +-
.../db/schemaengine/table/DataNodeTableCache.java | 2 +-
.../iotdb/db/storageengine/StorageEngine.java | 18 +-
.../db/storageengine/dataregion/DataRegion.java | 26 +-
.../compaction/execute/utils/CompactionUtils.java | 2 +-
.../memtable/AlignedWritableMemChunk.java | 92 +++-
.../dataregion/modification/DeletionPredicate.java | 46 +-
.../modification/TableDeletionEntry.java | 6 +-
.../{IDPredicate.java => TagPredicate.java} | 261 ++++++++--
.../dataregion/tsfile/TsFileResource.java | 41 +-
.../file/UnsealedTsFileRecoverPerformer.java | 2 +-
.../iotdb/db/metadata/path/PatternTreeMapTest.java | 2 +-
.../pipe/agent/task/PipeDataNodeTaskAgentTest.java | 88 ----
.../db/pipe/consensus/DeletionRecoverTest.java | 2 +-
.../db/pipe/consensus/DeletionResourceTest.java | 2 +-
.../pipe/event/TsFileInsertionEventParserTest.java | 154 ++++++
.../iotdb/db/pipe/sink/PipeReceiverTest.java | 70 ++-
.../PipeTabletEventPlainBatchPerformanceTest.java | 137 +++++
.../PipePlanTablePatternParseVisitorTest.java | 57 ++-
.../db/pipe/source/PipeRealtimeExtractTest.java | 73 +++
.../db/queryengine/common/MPPQueryContextTest.java | 125 +++++
.../queryengine/plan/analyze/AnalyzeUtilsTest.java | 6 +-
.../node/write/RelationalDeleteDataNodeTest.java | 8 +-
.../ExternalTsFileQueryResourceTest.java | 7 +-
.../schemaengine/table/DataNodeTableCacheTest.java | 20 +
.../iotdb/db/storageengine/StorageEngineTest.java | 45 ++
.../storageengine/dataregion/DataRegionTest.java | 2 +-
.../tablemodel/CompactionWithAllNullRowsTest.java | 12 +-
.../modification/ModificationFileTest.java | 27 +-
.../modification/TableDeletionEntryTest.java | 48 +-
.../dataregion/tsfile/TsFileResourceTest.java | 19 +-
.../conf/iotdb-system.properties.template | 2 +-
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 +
.../apache/iotdb/commons/i18n/UtilMessages.java | 2 +
.../apache/iotdb/commons/i18n/PipeMessages.java | 2 +
.../apache/iotdb/commons/i18n/UtilMessages.java | 2 +
.../apache/iotdb/commons/disk/FolderManager.java | 30 +-
.../MinFolderOccupiedSpaceFirstStrategy.java | 3 +-
.../pipe/config/constant/PipeSourceConstant.java | 41 ++
.../env/PipeTaskProcessorRuntimeEnvironment.java | 23 +
.../datastructure/visibility/VisibilityUtils.java | 9 +-
.../commons/pipe/receiver/IoTDBFileReceiver.java | 103 +++-
.../thrift/common/PipeTransferSliceReqHandler.java | 37 ++
.../thrift/request/PipeTransferCompressedReq.java | 21 +
.../iotdb/commons/pipe/source/IoTDBSource.java | 18 +-
.../apache/iotdb/commons/utils/JVMCommonUtils.java | 30 +-
...FolderOccupiedSpaceFirstStrategyRealFsTest.java | 61 +++
.../visibility/VisibilityUtilsTest.java | 24 +
.../pipe/receiver/IoTDBFileReceiverTest.java | 184 ++++++-
.../common/PipeTransferSliceReqBuilderTest.java | 52 ++
.../request/PipeTransferCompressedReqTest.java | 37 ++
.../iotdb/commons/utils/JVMCommonUtilsTest.java | 48 ++
133 files changed, 5226 insertions(+), 1160 deletions(-)
(iotdb) 02/02: Add RelationalInsertMultiTabletsNode
Posted by ji...@apache.org.
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 <jt...@163.com>
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;