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 54132275259 Pipe: optimize InsertRows schema aggregation (#18195)
(#18207)
54132275259 is described below
commit 54132275259e59bbeb295cf491b8de8d2f2603d1
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 15 09:48:41 2026 +0800
Pipe: optimize InsertRows schema aggregation (#18195) (#18207)
* Pipe: optimize InsertRows schema aggregation
* Pipe: add manual schema aggregation benchmark
* Pipe: optimize measurement subsequence aggregation
---
.../realtime/epoch/TsFileEpochManager.java | 69 +++++--
.../epoch/TsFileEpochManagerPerformanceTest.java | 210 +++++++++++++++++++++
.../realtime/epoch/TsFileEpochManagerTest.java | 109 +++++++++++
3 files changed, 375 insertions(+), 13 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java
index 003260ef1cc..8a7616cb1d5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManager.java
@@ -31,13 +31,15 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
public class TsFileEpochManager {
@@ -87,17 +89,58 @@ public class TsFileEpochManager {
event.getPipePattern());
}
- private Map<String, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
+ static Map<String, String[]> getDevice2MeasurementsMapFromInsertRowsNode(
InsertRowsNode insertRowsNode) {
- return insertRowsNode.getInsertRowNodeList().stream()
- .collect(
- Collectors.toMap(
- insertRowNode -> insertRowNode.getDevicePath().getFullPath(),
- InsertNode::getMeasurements,
- (oldMeasurements, newMeasurements) ->
- Stream.of(Arrays.asList(oldMeasurements),
Arrays.asList(newMeasurements))
- .flatMap(Collection::stream)
- .distinct()
- .toArray(String[]::new)));
+ final Map<String, String[]> device2Measurements = new HashMap<>();
+ final Map<String, Set<String>> device2DistinctMeasurements = new
HashMap<>();
+
+ // This method runs synchronously in the write path. Rebuilding a stream,
a hash set, and an
+ // array for every row is expensive when an InsertRowsNode contains many
rows for one device.
+ // Keep one set per repeated device instead and materialize its array only
once.
+ for (final InsertNode insertRowNode :
insertRowsNode.getInsertRowNodeList()) {
+ final String device = insertRowNode.getDevicePath().getFullPath();
+ final String[] measurements =
Objects.requireNonNull(insertRowNode.getMeasurements());
+ final String[] firstMeasurements =
device2Measurements.putIfAbsent(device, measurements);
+ if (firstMeasurements == null) {
+ continue;
+ }
+
+ final Set<String> distinctMeasurements =
+ device2DistinctMeasurements.computeIfAbsent(
+ device, key -> new
LinkedHashSet<>(Arrays.asList(firstMeasurements)));
+ addDistinctMeasurements(firstMeasurements, measurements,
distinctMeasurements);
+ }
+
+ device2DistinctMeasurements.forEach(
+ (device, measurements) ->
+ device2Measurements.put(device, measurements.toArray(new
String[measurements.size()])));
+ return device2Measurements;
+ }
+
+ private static void addDistinctMeasurements(
+ final String[] firstMeasurements,
+ final String[] measurements,
+ final Set<String> distinctMeasurements) {
+ if (measurements.length >= firstMeasurements.length) {
+ if (!Arrays.equals(firstMeasurements, measurements)) {
+ Collections.addAll(distinctMeasurements, measurements);
+ }
+ return;
+ }
+
+ // Skip the longest prefix that is already present as an ordered
subsequence of the first row.
+ int firstMeasurementIndex = 0;
+ int measurementIndex = 0;
+ while (firstMeasurementIndex < firstMeasurements.length
+ && measurementIndex < measurements.length) {
+ if (Objects.equals(
+ firstMeasurements[firstMeasurementIndex],
measurements[measurementIndex])) {
+ ++measurementIndex;
+ }
+ ++firstMeasurementIndex;
+ }
+ while (measurementIndex < measurements.length) {
+ distinctMeasurements.add(measurements[measurementIndex++]);
+ }
}
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerPerformanceTest.java
new file mode 100644
index 00000000000..cd47f23377a
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerPerformanceTest.java
@@ -0,0 +1,210 @@
+/*
+ * 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.source.dataregion.realtime.epoch;
+
+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.InsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
+
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Locale;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class TsFileEpochManagerPerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
+ "iotdb.pipe.epoch.insert.rows.aggregation.perf.enabled";
+ private static final String ROWS_PROPERTY =
"iotdb.pipe.epoch.insert.rows.aggregation.perf.rows";
+ private static final String MEASUREMENTS_PROPERTY =
+ "iotdb.pipe.epoch.insert.rows.aggregation.perf.measurements";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.pipe.epoch.insert.rows.aggregation.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
+ "iotdb.pipe.epoch.insert.rows.aggregation.perf.iterations";
+
+ private static volatile Map<String, String[]> benchmarkBlackhole;
+
+ @Test
+ public void sameDeviceSameSchemaAggregationBenchmark() {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s, -D%s and -D%s.",
+ ENABLED_PROPERTY,
+ ROWS_PROPERTY,
+ MEASUREMENTS_PROPERTY,
+ WARMUP_ITERATIONS_PROPERTY,
+ ITERATIONS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+
+ final int rowCount = Integer.getInteger(ROWS_PROPERTY, 10_000);
+ final int measurementCount = Integer.getInteger(MEASUREMENTS_PROPERTY,
100);
+ final int warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 5);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 10);
+ Assert.assertTrue(rowCount > 1);
+ Assert.assertTrue(measurementCount > 0);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+
+ final InsertRowsNode insertRowsNode = createInsertRowsNode(rowCount,
measurementCount);
+ assertSchemaInfoEquals(
+ getLegacyDevice2MeasurementsMap(insertRowsNode),
+
TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(insertRowsNode));
+
+ for (int i = 0; i < warmupIterations; ++i) {
+ if ((i & 1) == 0) {
+ benchmarkBlackhole = getLegacyDevice2MeasurementsMap(insertRowsNode);
+ benchmarkBlackhole =
+
TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(insertRowsNode);
+ } else {
+ benchmarkBlackhole =
+
TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(insertRowsNode);
+ benchmarkBlackhole = getLegacyDevice2MeasurementsMap(insertRowsNode);
+ }
+ }
+
+ final BenchmarkResult result = runBenchmark(insertRowsNode, iterations);
+ System.out.printf(
+ Locale.ROOT,
+ "InsertRows schema aggregation benchmark: rows=%d, measurements=%d,
warmups=%d, iterations=%d, legacy median=%.3f ms/op, optimized median=%.3f
ms/op, speedup=%.2fx%n",
+ rowCount,
+ measurementCount,
+ warmupIterations,
+ iterations,
+ toMillis(result.getLegacyMedianNanos()),
+ toMillis(result.getOptimizedMedianNanos()),
+ result.getLegacyMedianNanos() / result.getOptimizedMedianNanos());
+ }
+
+ private static BenchmarkResult runBenchmark(
+ final InsertRowsNode insertRowsNode, final int iterations) {
+ final long[] legacyElapsedNanos = new long[iterations];
+ final long[] optimizedElapsedNanos = new long[iterations];
+
+ for (int i = 0; i < iterations; ++i) {
+ if ((i & 1) == 0) {
+ legacyElapsedNanos[i] = measureLegacyAggregation(insertRowsNode);
+ optimizedElapsedNanos[i] = measureOptimizedAggregation(insertRowsNode);
+ } else {
+ optimizedElapsedNanos[i] = measureOptimizedAggregation(insertRowsNode);
+ legacyElapsedNanos[i] = measureLegacyAggregation(insertRowsNode);
+ }
+ }
+
+ return new BenchmarkResult(median(legacyElapsedNanos),
median(optimizedElapsedNanos));
+ }
+
+ private static long measureLegacyAggregation(final InsertRowsNode
insertRowsNode) {
+ final long startTime = System.nanoTime();
+ benchmarkBlackhole = getLegacyDevice2MeasurementsMap(insertRowsNode);
+ return System.nanoTime() - startTime;
+ }
+
+ private static long measureOptimizedAggregation(final InsertRowsNode
insertRowsNode) {
+ final long startTime = System.nanoTime();
+ benchmarkBlackhole =
+
TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(insertRowsNode);
+ return System.nanoTime() - startTime;
+ }
+
+ private static Map<String, String[]> getLegacyDevice2MeasurementsMap(
+ final InsertRowsNode insertRowsNode) {
+ // Keep the pre-optimization implementation from base commit 7ffb1364b84
for comparison.
+ return insertRowsNode.getInsertRowNodeList().stream()
+ .collect(
+ Collectors.toMap(
+ insertRowNode -> insertRowNode.getDevicePath().getFullPath(),
+ insertRowNode -> insertRowNode.getMeasurements(),
+ (oldMeasurements, newMeasurements) ->
+ Stream.of(Arrays.asList(oldMeasurements),
Arrays.asList(newMeasurements))
+ .flatMap(Collection::stream)
+ .distinct()
+ .toArray(String[]::new)));
+ }
+
+ private static InsertRowsNode createInsertRowsNode(
+ final int rowCount, final int measurementCount) {
+ final String[] measurements = new String[measurementCount];
+ for (int i = 0; i < measurementCount; ++i) {
+ measurements[i] = "s" + i;
+ }
+
+ final PlanNodeId planNodeId = new
PlanNodeId("schema-aggregation-performance");
+ final PartialPath devicePath = new
PartialPath("root.schema_aggregation_performance.d1", false);
+ final InsertRowsNode insertRowsNode = new InsertRowsNode(planNodeId);
+ for (int i = 0; i < rowCount; ++i) {
+ final InsertRowNode insertRowNode = new InsertRowNode(planNodeId);
+ insertRowNode.setDevicePath(devicePath);
+ // insertRecords creates a separate measurement array and decoded
strings for every row.
+ final String[] rowMeasurements = new String[measurementCount];
+ for (int j = 0; j < measurementCount; ++j) {
+ rowMeasurements[j] = new String(measurements[j].toCharArray());
+ }
+ insertRowNode.setMeasurements(rowMeasurements);
+ insertRowsNode.addOneInsertRowNode(insertRowNode, i);
+ }
+ return insertRowsNode;
+ }
+
+ private static void assertSchemaInfoEquals(
+ final Map<String, String[]> expected, final Map<String, String[]>
actual) {
+ Assert.assertEquals(expected.keySet(), actual.keySet());
+ expected.forEach(
+ (deviceID, measurements) -> Assert.assertArrayEquals(measurements,
actual.get(deviceID)));
+ }
+
+ 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;
+ }
+
+ private static double toMillis(final double nanos) {
+ return nanos / 1_000_000.0;
+ }
+
+ private static class BenchmarkResult {
+
+ private final double legacyMedianNanos;
+ private final double optimizedMedianNanos;
+
+ private BenchmarkResult(final double legacyMedianNanos, final double
optimizedMedianNanos) {
+ this.legacyMedianNanos = legacyMedianNanos;
+ this.optimizedMedianNanos = optimizedMedianNanos;
+ }
+
+ private double getLegacyMedianNanos() {
+ return legacyMedianNanos;
+ }
+
+ private double getOptimizedMedianNanos() {
+ return optimizedMedianNanos;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java
new file mode 100644
index 00000000000..4a11f2dec32
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/epoch/TsFileEpochManagerTest.java
@@ -0,0 +1,109 @@
+/*
+ * 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.source.dataregion.realtime.epoch;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import
org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+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.storageengine.dataregion.tsfile.TsFileResource;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class TsFileEpochManagerTest {
+
+ @Test
+ public void testMeasurementsAreDeduplicatedPerDevice() {
+ final String device = "root.test.d1";
+ final String uniqueDevice = "root.test.d2";
+ final InsertRowsNode rows = mock(InsertRowsNode.class);
+ final List<InsertRowNode> insertRowNodes =
+ Arrays.asList(
+ mockRow(device, "s1", null, "s2", "s1"),
+ mockRow(device, "s1", null, "s2", "s1"),
+ mockRow(device, "s2", "s3"),
+ mockRow(device, "s4", "s1"),
+ mockRow(uniqueDevice, "s5", "s5"));
+ when(rows.getInsertRowNodeList()).thenReturn(insertRowNodes);
+ when(rows.getMinTime()).thenReturn(1L);
+
+ final TsFileResource resource = mock(TsFileResource.class);
+ when(resource.getTsFilePath()).thenReturn("test.tsfile");
+ final PipeInsertNodeTabletInsertionEvent insertionEvent =
+ mock(PipeInsertNodeTabletInsertionEvent.class);
+ final PipeRealtimeEvent event =
+ new TsFileEpochManager()
+ .bindPipeInsertNodeTabletInsertionEvent(insertionEvent, rows,
resource);
+
+ Assert.assertEquals(2, event.getSchemaInfo().size());
+ Assert.assertArrayEquals(
+ new String[] {"s1", null, "s2", "s3", "s4"},
event.getSchemaInfo().get(device));
+ Assert.assertArrayEquals(new String[] {"s5", "s5"},
event.getSchemaInfo().get(uniqueDevice));
+ }
+
+ @Test
+ public void testSubsequenceAggregationPreservesEncounterOrder() {
+ final String device = "root.test.d1";
+ final InsertRowsNode rows = mock(InsertRowsNode.class);
+ final List<InsertRowNode> insertRowNodes =
+ Arrays.asList(
+ mockRow(device, "s1", "s3", "s1", null),
+ mockRow(device, copyString("s1"), copyString("s3"), null),
+ mockRow(device, copyString("s1"), "s2", copyString("s3")));
+ when(rows.getInsertRowNodeList()).thenReturn(insertRowNodes);
+
+ final Map<String, String[]> device2Measurements =
+ TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(rows);
+
+ Assert.assertArrayEquals(
+ new String[] {"s1", "s3", null, "s2"},
device2Measurements.get(device));
+ }
+
+ @Test(expected = NullPointerException.class)
+ public void testNullMeasurementsAreRejected() {
+ final String device = "root.test.d1";
+ final InsertRowsNode rows = mock(InsertRowsNode.class);
+ final List<InsertRowNode> insertRowNodes =
+ Arrays.asList(mockRow(device, "s1"), mockRow(device, (String[]) null));
+ when(rows.getInsertRowNodeList()).thenReturn(insertRowNodes);
+
+ TsFileEpochManager.getDevice2MeasurementsMapFromInsertRowsNode(rows);
+ }
+
+ private static InsertRowNode mockRow(final String device, final String...
measurements) {
+ final InsertRowNode row = mock(InsertRowNode.class);
+ when(row.getDevicePath()).thenReturn(new PartialPath(device, false));
+ when(row.getMeasurements()).thenReturn(measurements);
+ return row;
+ }
+
+ private static String copyString(final String value) {
+ return new String(value.toCharArray());
+ }
+}
