This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 780540cb87a [To dev/1.3] Optimize pipe tablet memory accounting
(#18211) (#18220)
780540cb87a is described below
commit 780540cb87af7c1509964ac5f0411adaaafc4794
Author: Caideyipi <[email protected]>
AuthorDate: Thu Jul 16 11:08:32 2026 +0800
[To dev/1.3] Optimize pipe tablet memory accounting (#18211) (#18220)
---
.../resource/memory/InsertNodeMemoryEstimator.java | 17 ++-
.../request/PipeTransferTabletBatchReq.java | 11 +-
.../InsertNodeMemoryEstimatorPerformanceTest.java | 104 +++++++++++++++++
.../memory/InsertNodeMemoryEstimatorTest.java | 79 +++++++++++++
.../pipe/sink/PipeDataNodeThriftRequestTest.java | 16 +++
.../PipeTransferTabletBatchReqPerformanceTest.java | 126 +++++++++++++++++++++
6 files changed, 343 insertions(+), 10 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 9616777b021..b528850f276 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
@@ -190,7 +190,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertTabletNode(final InsertTabletNode node) {
- return sizeOfInsertTabletNode(node, newDeduplicatedObjectSet());
+ return sizeOfInsertTabletNode(node, null);
}
private static long sizeOfInsertTabletNode(
@@ -205,7 +205,7 @@ public class InsertNodeMemoryEstimator {
}
private static long sizeOfInsertRowNode(final InsertRowNode node) {
- return sizeOfInsertRowNode(node, newDeduplicatedObjectSet());
+ return sizeOfInsertRowNode(node, null);
}
private static long sizeOfInsertRowNode(
@@ -217,7 +217,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);
@@ -227,7 +228,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);
@@ -237,7 +239,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);
@@ -705,6 +708,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/PipeTransferTabletBatchReq.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReq.java
index 266894060dc..289aff117c9 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReq.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReq.java
@@ -130,20 +130,21 @@ public class PipeTransferTabletBatchReq extends
TPipeTransferReq {
public static PipeTransferTabletBatchReq fromTPipeTransferReq(
final TPipeTransferReq transferReq) {
final PipeTransferTabletBatchReq batchReq = new
PipeTransferTabletBatchReq();
- final TabletStringInternPool tabletStringInternPool = new
TabletStringInternPool();
// Binary req, for rolling upgrading
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(
PipeTransferTabletInsertNodeReq.toTPipeTransferRawReq(
(InsertNode) PlanFragment.deserializeHelper(transferReq.body,
null)));
}
- 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(
PipeTransferTabletRawReq.toTPipeTransferRawReq(transferReq.body,
tabletStringInternPool));
}
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..1a4fc3f9e8e
--- /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.db.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 58b2756b16f..c1e715c7d1f 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
@@ -24,6 +24,7 @@ import
org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
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;
@@ -150,6 +151,84 @@ 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(
+ createTextInsertTabletNode("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.getMeasurementId(),
firstMeasurementSchema.getType());
+ }
+
private static InsertRowsNode createInsertRowsNode(
String planNodeId, InsertRowNode... insertRowNodes) {
InsertRowsNode node = new InsertRowsNode(new PlanNodeId(planNodeId));
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 1dae3b55f97..95f1cdd146c 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
@@ -359,6 +359,22 @@ public class PipeDataNodeThriftRequestTest {
Assert.assertFalse(deserializedReq.getTabletReqs().get(0).getIsAligned());
}
+ @Test
+ public void
testPipeTransferTabletBatchReqSkipsStringInterningForSingleRawTablet()
+ throws IOException {
+ final PipeTransferTabletBatchReq deserializedReq =
+ PipeTransferTabletBatchReq.fromTPipeTransferReq(
+ PipeTransferTabletBatchReq.toTPipeTransferReq(
+ Collections.emptyList(),
+ Collections.singletonList(
+ serializeTablet(
+ createSingleValueTablet(new String("s1"), new
String("s1")), false))));
+
+ final Tablet tablet = deserializedReq.getTabletReqs().get(0).getTablet();
+ Assert.assertEquals(tablet.deviceId,
tablet.getSchemas().get(0).getMeasurementId());
+ Assert.assertNotSame(tablet.deviceId,
tablet.getSchemas().get(0).getMeasurementId());
+ }
+
@Test
public void testPipeTransferTabletBatchReqInternsRepeatedMeasurementNames()
throws IOException {
final List<ByteBuffer> tabletBuffers = new ArrayList<>();
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqPerformanceTest.java
new file mode 100644
index 00000000000..69473573b53
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/payload/evolvable/request/PipeTransferTabletBatchReqPerformanceTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.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 PipeTransferTabletBatchReqPerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.pipe.tablet.batch.deserialize.perf.enabled";
+ private static final String COLUMNS_PROPERTY =
"iotdb.pipe.tablet.batch.deserialize.perf.columns";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.pipe.tablet.batch.deserialize.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
+ "iotdb.pipe.tablet.batch.deserialize.perf.iterations";
+
+ private static volatile PipeTransferTabletBatchReq 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 PipeTransferTabletBatchReq request =
createSingleWideRawTabletBatch(columnCount);
+ final PipeTransferTabletBatchReq 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 V1 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 PipeTransferTabletBatchReq
createSingleWideRawTabletBatch(final int columnCount)
+ throws IOException {
+ final List<MeasurementSchema> 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 PipeTransferTabletBatchReq.toTPipeTransferReq(
+ Collections.emptyList(),
Collections.singletonList(serializeTablet(tablet)));
+ }
+
+ 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 PipeTransferTabletBatchReq deserialize(final
PipeTransferTabletBatchReq request) {
+ request.body.position(0);
+ benchmarkBlackhole =
PipeTransferTabletBatchReq.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;
+ }
+}
