This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new fe1777bfa6d Optimize pipe tablet memory accounting (#18211)
fe1777bfa6d is described below
commit fe1777bfa6d54f6847e910a9f5aebdaaf5a103e2
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 15 17:14:19 2026 +0800
Optimize pipe tablet memory accounting (#18211)
---
.../resource/memory/InsertNodeMemoryEstimator.java | 24 ++--
.../request/PipeTransferTabletBatchReqV2.java | 13 +-
.../request/PipeTransferTabletRawReqV2.java | 16 +--
.../InsertNodeMemoryEstimatorPerformanceTest.java | 104 ++++++++++++++++
.../memory/InsertNodeMemoryEstimatorTest.java | 130 ++++++++++++++++++++
.../pipe/sink/PipeDataNodeThriftRequestTest.java | 45 +++++++
...ipeTransferTabletBatchReqV2PerformanceTest.java | 133 +++++++++++++++++++++
7 files changed, 444 insertions(+), 21 deletions(-)
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..b7b660f638c 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
@@ -220,7 +220,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertTabletNode(final InsertTabletNode node) {
- return sizeOfInsertTabletNode(node, newDeduplicatedObjectSet());
+ return sizeOfInsertTabletNode(node, null);
}
private static long sizeOfInsertTabletNode(
@@ -235,7 +235,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowNode(final InsertRowNode node) {
- return sizeOfInsertRowNode(node, newDeduplicatedObjectSet());
+ return sizeOfInsertRowNode(node, null);
}
private static long sizeOfInsertRowNode(
@@ -247,7 +247,8 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowsNode(final InsertRowsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
+ final Set<Object> deduplicatedObjects =
+ newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
long size = INSERT_ROWS_NODE_SIZE;
size += calculateFullInsertNodeSize(node, deduplicatedObjects);
size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
@@ -257,7 +258,8 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowsOfOneDeviceNode(final
InsertRowsOfOneDeviceNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
+ final Set<Object> deduplicatedObjects =
+ newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
long size = INSERT_ROWS_OF_ONE_DEVICE_NODE_SIZE;
size += calculateFullInsertNodeSize(node, deduplicatedObjects);
size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
@@ -267,7 +269,8 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertMultiTabletsNode(final
InsertMultiTabletsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
+ final Set<Object> deduplicatedObjects =
+ newDeduplicatedObjectSetIfNeeded(node.getInsertTabletNodeList());
long size = INSERT_MULTI_TABLETS_NODE_SIZE;
size += calculateFullInsertNodeSize(node, deduplicatedObjects);
size += sizeOfInsertTabletNodeList(node.getInsertTabletNodeList(),
deduplicatedObjects);
@@ -277,7 +280,8 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfRelationalInsertRowsNode(final
RelationalInsertRowsNode node) {
- final Set<Object> deduplicatedObjects = newDeduplicatedObjectSet();
+ final Set<Object> deduplicatedObjects =
+ newDeduplicatedObjectSetIfNeeded(node.getInsertRowNodeList());
long size = RELATIONAL_INSERT_ROWS_NODE_SIZE;
size += calculateFullInsertNodeSize(node, deduplicatedObjects);
size += sizeOfInsertRowNodeList(node.getInsertRowNodeList(),
deduplicatedObjects);
@@ -287,7 +291,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfRelationalInsertRowNode(final
RelationalInsertRowNode node) {
- return sizeOfRelationalInsertRowNode(node, newDeduplicatedObjectSet());
+ return sizeOfRelationalInsertRowNode(node, null);
}
private static long sizeOfRelationalInsertRowNode(
@@ -299,7 +303,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfRelationalInsertTabletNode(final
RelationalInsertTabletNode node) {
- return sizeOfRelationalInsertTabletNode(node, newDeduplicatedObjectSet());
+ return sizeOfRelationalInsertTabletNode(node, null);
}
private static long sizeOfRelationalInsertTabletNode(
@@ -771,6 +775,10 @@ public class InsertNodeMemoryEstimator {
return Collections.newSetFromMap(new IdentityHashMap<>());
}
+ private static Set<Object> newDeduplicatedObjectSetIfNeeded(final List<?>
children) {
+ return children != null && children.size() > 1 ?
newDeduplicatedObjectSet() : null;
+ }
+
private static boolean shouldCountObject(
final Object object, final Set<Object> deduplicatedObjects) {
return object != null && (deduplicatedObjects == null ||
deduplicatedObjects.add(object));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java
index 6c4607518b4..4dd044cf273 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2.java
@@ -215,21 +215,22 @@ public class PipeTransferTabletBatchReqV2 extends
TPipeTransferReq {
public static PipeTransferTabletBatchReqV2 fromTPipeTransferReq(
final org.apache.iotdb.service.rpc.thrift.TPipeTransferReq transferReq) {
final PipeTransferTabletBatchReqV2 batchReq = new
PipeTransferTabletBatchReqV2();
- final TabletStringInternPool tabletStringInternPool = new
TabletStringInternPool();
// Binary req, for rolling upgrade
ReadWriteIOUtils.readInt(transferReq.body);
- int size = ReadWriteIOUtils.readInt(transferReq.body);
- for (int i = 0; i < size; ++i) {
+ final int insertNodeCount = ReadWriteIOUtils.readInt(transferReq.body);
+ for (int i = 0; i < insertNodeCount; ++i) {
batchReq.insertNodeReqs.add(
PipeTransferTabletInsertNodeReqV2.toTabletInsertNodeReq(
(InsertNode) PlanFragment.deserializeHelper(transferReq.body,
null),
-
tabletStringInternPool.intern(ReadWriteIOUtils.readString(transferReq.body))));
+ ReadWriteIOUtils.readString(transferReq.body)));
}
- size = ReadWriteIOUtils.readInt(transferReq.body);
- for (int i = 0; i < size; ++i) {
+ final int rawTabletCount = ReadWriteIOUtils.readInt(transferReq.body);
+ final TabletStringInternPool tabletStringInternPool =
+ rawTabletCount > 1 ? new TabletStringInternPool() : null;
+ for (int i = 0; i < rawTabletCount; ++i) {
batchReq.tabletReqs.add(
PipeTransferTabletRawReqV2.toTPipeTransferRawReq(
transferReq.body, tabletStringInternPool));
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java
index d395bf6cf5f..3537d9ddd00 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletRawReqV2.java
@@ -214,15 +214,12 @@ public class PipeTransferTabletRawReqV2 extends
PipeTransferTabletRawReq {
public void deserializeTPipeTransferRawReq(
final ByteBuffer buffer, final TabletStringInternPool
tabletStringInternPool) {
- final TabletStringInternPool internPool =
- Objects.nonNull(tabletStringInternPool)
- ? tabletStringInternPool
- : new TabletStringInternPool();
final int startPosition = buffer.position();
try {
// V2: read databaseName, readDatabaseName = true
final InsertTabletStatement insertTabletStatement =
-
TabletStatementConverter.deserializeStatementFromTabletFormat(buffer, true,
internPool);
+ TabletStatementConverter.deserializeStatementFromTabletFormat(
+ buffer, true, tabletStringInternPool);
this.isAligned = insertTabletStatement.isAligned();
// databaseName is already set in deserializeStatementFromTabletFormat
when
// readDatabaseName=true
@@ -232,9 +229,14 @@ public class PipeTransferTabletRawReqV2 extends
PipeTransferTabletRawReq {
// If Statement deserialization fails, fallback to Tablet format
// Reset buffer position for Tablet deserialization
buffer.position(startPosition);
- this.tablet = PipeTabletUtils.internTablet(Tablet.deserialize(buffer),
internPool);
+ this.tablet =
+ PipeTabletUtils.internTablet(Tablet.deserialize(buffer),
tabletStringInternPool);
this.isAligned = ReadWriteIOUtils.readBool(buffer);
- this.dataBaseName =
internPool.intern(ReadWriteIOUtils.readString(buffer));
+ final String dataBaseName = ReadWriteIOUtils.readString(buffer);
+ this.dataBaseName =
+ Objects.nonNull(tabletStringInternPool)
+ ? tabletStringInternPool.intern(dataBaseName)
+ : dataBaseName;
}
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
new file mode 100644
index 00000000000..d7c01a5b676
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorPerformanceTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.pipe.resource.memory;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+public class InsertNodeMemoryEstimatorPerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.perf.enabled";
+ private static final String COLUMNS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.perf.columns";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
+ "iotdb.pipe.insert.node.memory.estimator.perf.iterations";
+
+ private static volatile long benchmarkBlackhole;
+
+ @Test
+ public void wideTabletSizeOfBenchmark() throws IllegalPathException {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s and -D%s.",
+ ENABLED_PROPERTY, COLUMNS_PROPERTY, WARMUP_ITERATIONS_PROPERTY,
ITERATIONS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+
+ final int columnCount = Integer.getInteger(COLUMNS_PROPERTY, 100_000);
+ final int warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 3);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 10);
+ Assert.assertTrue(columnCount > 0);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+
+ final InsertTabletNode insertTabletNode =
createWideInsertTabletNode(columnCount);
+ runBenchmark(insertTabletNode, warmupIterations);
+ final long elapsedNanos = runBenchmark(insertTabletNode, iterations);
+
+ Assert.assertTrue(benchmarkBlackhole > 0);
+ System.out.printf(
+ "InsertNode memory estimate benchmark: columns=%d, warmups=%d,
iterations=%d, %.3f ms/op%n",
+ columnCount,
+ warmupIterations,
+ iterations,
+ elapsedNanos / (double) iterations / 1_000_000.0);
+ }
+
+ private static long runBenchmark(final InsertTabletNode insertTabletNode,
final int iterations) {
+ final long startTime = System.nanoTime();
+ for (int i = 0; i < iterations; ++i) {
+ benchmarkBlackhole = InsertNodeMemoryEstimator.sizeOf(insertTabletNode);
+ }
+ return System.nanoTime() - startTime;
+ }
+
+ private static InsertTabletNode createWideInsertTabletNode(final int
columnCount)
+ throws IllegalPathException {
+ final String[] measurements = new String[columnCount];
+ final TSDataType[] dataTypes = new TSDataType[columnCount];
+ final MeasurementSchema[] measurementSchemas = new
MeasurementSchema[columnCount];
+ for (int i = 0; i < columnCount; ++i) {
+ measurements[i] = "s" + i;
+ dataTypes[i] = TSDataType.INT32;
+ measurementSchemas[i] = new MeasurementSchema(measurements[i],
TSDataType.INT32);
+ }
+ return new InsertTabletNode(
+ new PlanNodeId("wide-memory-estimator"),
+ new PartialPath("root.sg.d1"),
+ false,
+ measurements,
+ dataTypes,
+ measurementSchemas,
+ new long[] {0L},
+ null,
+ new Object[columnCount],
+ 1);
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorTest.java
index 06db08aaf57..ad809bd4abb 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/resource/memory/InsertNodeMemoryEstimatorTest.java
@@ -23,11 +23,15 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+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.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.InsertRowsOfOneDeviceNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.common.conf.TSFileConfig;
@@ -162,6 +166,89 @@ public class InsertNodeMemoryEstimatorTest {
> InsertNodeMemoryEstimator.sizeOf(shortPlanNodeIdRow));
}
+ @Test
+ public void testLeafNodesDoNotDeduplicateRepeatedMeasurementSchemas()
+ throws IllegalPathException {
+ assertLeafNodeDoesNotDeduplicateRepeatedMeasurementSchemas(
+ createTextInsertRowNode(
+ "row", "root.sg.d1", new String[] {"s0", "s1"}, new String[]
{"v0", "v1"}));
+ assertLeafNodeDoesNotDeduplicateRepeatedMeasurementSchemas(
+ createTextRelationalInsertRowNode(
+ "relational-row", "root.sg.d1", new String[] {"s0", "s1"}, new
String[] {"v0", "v1"}));
+ assertLeafNodeDoesNotDeduplicateRepeatedMeasurementSchemas(
+ createTextInsertTabletNode("tablet", "root.sg.d1", 2, 1, 1));
+ assertLeafNodeDoesNotDeduplicateRepeatedMeasurementSchemas(
+ createTextRelationalInsertTabletNode("relational-tablet",
"root.sg.d1", 2, 1, 1));
+ }
+
+ @Test
+ public void
testSingleChildCompositesDoNotDeduplicateRepeatedMeasurementSchemas()
+ throws IllegalPathException {
+ final InsertRowNode rowWithDistinctSchemas =
+ createTextInsertRowNode(
+ "row", "root.sg.d1", new String[] {"s0", "s1"}, new String[]
{"v0", "v1"});
+ final InsertRowNode rowWithRepeatedSchema =
+ createTextInsertRowNode(
+ "row", "root.sg.d1", new String[] {"s0", "s1"}, new String[]
{"v0", "v1"});
+ replaceSecondSchemaWithEquivalentDistinctSchema(rowWithDistinctSchemas);
+ rowWithRepeatedSchema.getMeasurementSchemas()[1] =
+ rowWithRepeatedSchema.getMeasurementSchemas()[0];
+ Assert.assertEquals(
+ InsertNodeMemoryEstimator.sizeOf(createInsertRowsNode("parent",
rowWithDistinctSchemas)),
+ InsertNodeMemoryEstimator.sizeOf(createInsertRowsNode("parent",
rowWithRepeatedSchema)));
+
+ final InsertTabletNode tabletWithDistinctSchemas =
+ createTextInsertTabletNode("tablet", "root.sg.d1", 2, 1, 1);
+ final InsertTabletNode tabletWithRepeatedSchema =
+ createTextInsertTabletNode("tablet", "root.sg.d1", 2, 1, 1);
+ replaceSecondSchemaWithEquivalentDistinctSchema(tabletWithDistinctSchemas);
+ tabletWithRepeatedSchema.getMeasurementSchemas()[1] =
+ tabletWithRepeatedSchema.getMeasurementSchemas()[0];
+ Assert.assertEquals(
+ InsertNodeMemoryEstimator.sizeOf(
+ createInsertMultiTabletsNode("parent", tabletWithDistinctSchemas)),
+ InsertNodeMemoryEstimator.sizeOf(
+ createInsertMultiTabletsNode("parent", tabletWithRepeatedSchema)));
+ }
+
+ @Test
+ public void testMultiChildCompositeDeduplicatesSharedMeasurementSchemas()
+ throws IllegalPathException {
+ final long sizeWithDistinctSchemas =
+ InsertNodeMemoryEstimator.sizeOf(
+ createInsertMultiTabletsNode(
+ "parent",
+ createTextInsertTabletNode("tablet-1", "root.sg.d1", 2, 1, 1),
+ createTextInsertTabletNode("tablet-2", "root.sg.d2", 2, 1,
1)));
+
+ final InsertTabletNode firstTablet =
+ createTextInsertTabletNode("tablet-1", "root.sg.d1", 2, 1, 1);
+ final InsertTabletNode secondTablet =
+ createTextInsertTabletNode("tablet-2", "root.sg.d2", 2, 1, 1);
+ secondTablet.setMeasurementSchemas(firstTablet.getMeasurementSchemas());
+
+ Assert.assertTrue(
+ InsertNodeMemoryEstimator.sizeOf(
+ createInsertMultiTabletsNode("parent", firstTablet,
secondTablet))
+ < sizeWithDistinctSchemas);
+ }
+
+ private static void
assertLeafNodeDoesNotDeduplicateRepeatedMeasurementSchemas(
+ final InsertNode node) {
+ replaceSecondSchemaWithEquivalentDistinctSchema(node);
+ final long sizeWithDistinctSchemas =
InsertNodeMemoryEstimator.sizeOf(node);
+ node.getMeasurementSchemas()[1] = node.getMeasurementSchemas()[0];
+
+ Assert.assertEquals(sizeWithDistinctSchemas,
InsertNodeMemoryEstimator.sizeOf(node));
+ }
+
+ private static void replaceSecondSchemaWithEquivalentDistinctSchema(final
InsertNode node) {
+ final MeasurementSchema firstMeasurementSchema =
node.getMeasurementSchemas()[0];
+ node.getMeasurementSchemas()[1] =
+ new MeasurementSchema(
+ firstMeasurementSchema.getMeasurementName(),
firstMeasurementSchema.getType());
+ }
+
private static InsertRowsNode createInsertRowsNode(
String planNodeId, InsertRowNode... insertRowNodes) {
InsertRowsNode node = new InsertRowsNode(new PlanNodeId(planNodeId));
@@ -220,6 +307,24 @@ public class InsertNodeMemoryEstimatorTest {
false);
}
+ private static RelationalInsertRowNode createTextRelationalInsertRowNode(
+ String planNodeId, String devicePath, String[] measurements, String[]
values)
+ throws IllegalPathException {
+ final InsertRowNode insertRowNode =
+ createTextInsertRowNode(planNodeId, devicePath, measurements, values);
+ return new RelationalInsertRowNode(
+ new PlanNodeId(planNodeId),
+ new PartialPath(devicePath),
+ false,
+ insertRowNode.getMeasurements(),
+ insertRowNode.getDataTypes(),
+ insertRowNode.getMeasurementSchemas(),
+ 1L,
+ insertRowNode.getValues(),
+ false,
+ createFieldColumnCategories(measurements.length));
+ }
+
private static InsertTabletNode createTextInsertTabletNode(
String planNodeId, String devicePath, int measurementCount, int
rowCount, int repeatCount)
throws IllegalPathException {
@@ -260,6 +365,31 @@ public class InsertNodeMemoryEstimatorTest {
rowCount);
}
+ private static RelationalInsertTabletNode
createTextRelationalInsertTabletNode(
+ String planNodeId, String devicePath, int measurementCount, int
rowCount, int repeatCount)
+ throws IllegalPathException {
+ final InsertTabletNode insertTabletNode =
+ createTextInsertTabletNode(planNodeId, devicePath, measurementCount,
rowCount, repeatCount);
+ return new RelationalInsertTabletNode(
+ new PlanNodeId(planNodeId),
+ new PartialPath(devicePath),
+ false,
+ insertTabletNode.getMeasurements(),
+ insertTabletNode.getDataTypes(),
+ insertTabletNode.getMeasurementSchemas(),
+ insertTabletNode.getTimes(),
+ insertTabletNode.getBitMaps(),
+ insertTabletNode.getColumns(),
+ insertTabletNode.getRowCount(),
+ createFieldColumnCategories(measurementCount));
+ }
+
+ private static TsTableColumnCategory[] createFieldColumnCategories(final int
measurementCount) {
+ final TsTableColumnCategory[] columnCategories = new
TsTableColumnCategory[measurementCount];
+ Arrays.fill(columnCategories, TsTableColumnCategory.FIELD);
+ return columnCategories;
+ }
+
private static TSStatus createStatus(String message) {
TSStatus status = new TSStatus();
status.setCode(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java
index ab08dc6b96e..07b7661e2cb 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/PipeDataNodeThriftRequestTest.java
@@ -911,6 +911,51 @@ public class PipeDataNodeThriftRequestTest {
Assert.assertEquals("test",
deserializedReq.getInsertNodeReqs().get(0).getDataBaseName());
}
+ @Test
+ public void
testPipeTransferTabletBatchReqV2SkipsStringInterningForSingleRawTablet()
+ throws IOException {
+ final PipeTransferTabletBatchReqV2 deserializedReq =
+ PipeTransferTabletBatchReqV2.fromTPipeTransferReq(
+ PipeTransferTabletBatchReqV2.toTPipeTransferReq(
+ Collections.emptyList(),
+ Collections.singletonList(
+ serializeTablet(
+ createSingleValueTablet(new String("root.sg.d"), new
String("s1")), false)),
+ Collections.emptyList(),
+ Collections.singletonList(new String("root.sg.d"))));
+
+ final PipeTransferTabletRawReqV2 tabletReq =
deserializedReq.getTabletReqs().get(0);
+ Assert.assertEquals(tabletReq.getTablet().getDeviceId(),
tabletReq.getDataBaseName());
+ Assert.assertNotSame(tabletReq.getTablet().getDeviceId(),
tabletReq.getDataBaseName());
+ }
+
+ @Test
+ public void
testPipeTransferTabletBatchReqV2InternsStringsAcrossMultipleRawTablets()
+ throws IOException {
+ final List<ByteBuffer> tabletBuffers = new ArrayList<>();
+ tabletBuffers.add(
+ serializeTablet(createSingleValueTablet(new String("root.sg.d"), new
String("s1")), false));
+ tabletBuffers.add(
+ serializeTablet(createSingleValueTablet(new String("root.sg.d"), new
String("s1")), false));
+
+ final PipeTransferTabletBatchReqV2 deserializedReq =
+ PipeTransferTabletBatchReqV2.fromTPipeTransferReq(
+ PipeTransferTabletBatchReqV2.toTPipeTransferReq(
+ Collections.emptyList(),
+ tabletBuffers,
+ Collections.emptyList(),
+ Arrays.asList(new String("root.sg.d"), new
String("root.sg.d"))));
+
+ final PipeTransferTabletRawReqV2 firstTabletReq =
deserializedReq.getTabletReqs().get(0);
+ final PipeTransferTabletRawReqV2 secondTabletReq =
deserializedReq.getTabletReqs().get(1);
+ Assert.assertSame(
+ firstTabletReq.getTablet().getDeviceId(),
secondTabletReq.getTablet().getDeviceId());
+ Assert.assertSame(firstTabletReq.getDataBaseName(),
secondTabletReq.getDataBaseName());
+ Assert.assertSame(
+ ((MeasurementSchema)
firstTabletReq.getTablet().getSchemas().get(0)).getMeasurementName(),
+ ((MeasurementSchema)
secondTabletReq.getTablet().getSchemas().get(0)).getMeasurementName());
+ }
+
@Test
public void testPipeTransferTabletBatchReqV2WithMultipleTreeModelDatabases()
throws IOException {
final List<ByteBuffer> insertNodeBuffers = new ArrayList<>();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2PerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2PerformanceTest.java
new file mode 100644
index 00000000000..96be39ba21e
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqV2PerformanceTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.pipe.sink.payload.evolvable.request;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.PublicBAOS;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.apache.tsfile.write.record.Tablet;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+
+public class PipeTransferTabletBatchReqV2PerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
+ "iotdb.pipe.tablet.batch.v2.deserialize.perf.enabled";
+ private static final String COLUMNS_PROPERTY =
+ "iotdb.pipe.tablet.batch.v2.deserialize.perf.columns";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.pipe.tablet.batch.v2.deserialize.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
+ "iotdb.pipe.tablet.batch.v2.deserialize.perf.iterations";
+
+ private static volatile PipeTransferTabletBatchReqV2 benchmarkBlackhole;
+
+ @Test
+ public void singleWideRawTabletDeserializationBenchmark() throws IOException
{
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s and -D%s.",
+ ENABLED_PROPERTY, COLUMNS_PROPERTY, WARMUP_ITERATIONS_PROPERTY,
ITERATIONS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+
+ final int columnCount = Integer.getInteger(COLUMNS_PROPERTY, 100_000);
+ final int warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 1);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 5);
+ Assert.assertTrue(columnCount > 0);
+ Assert.assertTrue(warmupIterations >= 0);
+ Assert.assertTrue(iterations > 0);
+
+ final PipeTransferTabletBatchReqV2 request =
createSingleWideRawTabletBatch(columnCount);
+ final PipeTransferTabletBatchReqV2 verificationResult =
deserialize(request);
+ Assert.assertEquals(1, verificationResult.getTabletReqs().size());
+ Assert.assertEquals(
+ columnCount,
verificationResult.getTabletReqs().get(0).getTablet().getSchemas().size());
+
+ for (int i = 0; i < warmupIterations; ++i) {
+ deserialize(request);
+ }
+
+ final long[] elapsedNanos = new long[iterations];
+ for (int i = 0; i < iterations; ++i) {
+ final long startTime = System.nanoTime();
+ deserialize(request);
+ elapsedNanos[i] = System.nanoTime() - startTime;
+ }
+
+ System.out.printf(
+ Locale.ROOT,
+ "Batch V2 single raw tablet deserialization benchmark: columns=%d,
warmups=%d, iterations=%d, median=%.3f ms/op%n",
+ columnCount,
+ warmupIterations,
+ iterations,
+ median(elapsedNanos) / 1_000_000.0);
+ }
+
+ private static PipeTransferTabletBatchReqV2
createSingleWideRawTabletBatch(final int columnCount)
+ throws IOException {
+ final List<IMeasurementSchema> schemas = new ArrayList<>(columnCount);
+ for (int i = 0; i < columnCount; ++i) {
+ schemas.add(new MeasurementSchema("s" + i, TSDataType.INT32));
+ }
+
+ final Tablet tablet = new Tablet("root.sg.d", schemas, 1);
+ return PipeTransferTabletBatchReqV2.toTPipeTransferReq(
+ Collections.emptyList(),
+ Collections.singletonList(serializeTablet(tablet)),
+ Collections.emptyList(),
+ Collections.singletonList("root.sg"));
+ }
+
+ private static ByteBuffer serializeTablet(final Tablet tablet) throws
IOException {
+ try (final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ final DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream)) {
+ tablet.serialize(outputStream);
+ ReadWriteIOUtils.write(false, outputStream);
+ return ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
+ }
+ }
+
+ private static PipeTransferTabletBatchReqV2 deserialize(
+ final PipeTransferTabletBatchReqV2 request) {
+ request.body.position(0);
+ benchmarkBlackhole =
PipeTransferTabletBatchReqV2.fromTPipeTransferReq(request);
+ return benchmarkBlackhole;
+ }
+
+ private static double median(final long[] values) {
+ Arrays.sort(values);
+ final int middle = values.length / 2;
+ return (values.length & 1) == 1
+ ? values[middle]
+ : values[middle - 1] + (values[middle] - values[middle - 1]) / 2.0;
+ }
+}
