This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch ancient-coding in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 5b6bf57519140271771d6aaf7f2b4f45a2e8f307 Author: Caideyipi <[email protected]> AuthorDate: Tue Jul 14 15:08:40 2026 +0800 Pipe: optimize InsertRows schema aggregation (#18195) * 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()); + } +}
