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 9e40726175d Reuse tablet device ranges during writes (#18734)
9e40726175d is described below
commit 9e40726175d4305bdaf5829f6b9725778a07d422
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 29 14:44:04 2026 +0800
Reuse tablet device ranges during writes (#18734)
---
.../dataregion/memtable/AbstractMemTable.java | 32 ++-
.../dataregion/memtable/IMemTable.java | 10 +
.../dataregion/memtable/TsFileProcessor.java | 80 +++++-
.../TabletDeviceRangeReusePerformanceTest.java | 283 +++++++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 57 +++++
5 files changed, 448 insertions(+), 14 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
index e24308cf505..35a3696630e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java
@@ -315,10 +315,26 @@ public abstract class AbstractMemTable implements
IMemTable {
public int insertAlignedTablet(
InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results)
throws WriteProcessException {
+ try {
+ return insertAlignedTablet(
+ insertTabletNode, start, end, results,
insertTabletNode.splitByDevice(start, end));
+ } catch (RuntimeException e) {
+ throw new WriteProcessException(e);
+ }
+ }
+
+ @Override
+ public int insertAlignedTablet(
+ InsertTabletNode insertTabletNode,
+ int start,
+ int end,
+ TSStatus[] results,
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs)
+ throws WriteProcessException {
try {
int nullPointsNumber =
computeTabletNullPointsNumber(insertTabletNode, start, end, true,
results);
- writeAlignedTablet(insertTabletNode, start, end, results);
+ writeAlignedTablet(insertTabletNode, start, end, results,
deviceEndOffsetPairs);
// TODO-Table: what is the relation between this and
TsFileProcessor.checkMemCost
memSize += MemUtils.getAlignedTabletSize(insertTabletNode, start, end,
results);
int validValueCount =
@@ -449,6 +465,16 @@ public abstract class AbstractMemTable implements
IMemTable {
public void writeAlignedTablet(
InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results) {
+ writeAlignedTablet(
+ insertTabletNode, start, end, results,
insertTabletNode.splitByDevice(start, end));
+ }
+
+ public void writeAlignedTablet(
+ InsertTabletNode insertTabletNode,
+ int start,
+ int end,
+ TSStatus[] results,
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs) {
List<IMeasurementSchema> schemaList = new ArrayList<>();
final int measurementSize =
insertTabletNode.getMeasurements() == null ? 0 :
insertTabletNode.getMeasurements().length;
@@ -474,10 +500,8 @@ public abstract class AbstractMemTable implements
IMemTable {
if (schemaList.isEmpty()) {
return;
}
- final List<Pair<IDeviceID, Integer>> deviceEndOffsetPair =
- insertTabletNode.splitByDevice(start, end);
int splitStart = start;
- for (Pair<IDeviceID, Integer> pair : deviceEndOffsetPair) {
+ for (Pair<IDeviceID, Integer> pair : deviceEndOffsetPairs) {
final IDeviceID deviceID = pair.left;
int splitEnd = pair.right;
IWritableMemChunkGroup memChunkGroup =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
index 4d38fea2c11..afa24083ff6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/IMemTable.java
@@ -123,6 +123,16 @@ public interface IMemTable extends WALEntryValue {
int insertAlignedTablet(InsertTabletNode insertTabletNode, int start, int
end, TSStatus[] results)
throws WriteProcessException;
+ default int insertAlignedTablet(
+ InsertTabletNode insertTabletNode,
+ int start,
+ int end,
+ TSStatus[] results,
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs)
+ throws WriteProcessException {
+ return insertAlignedTablet(insertTabletNode, start, end, results);
+ }
+
ReadOnlyMemChunk query(
QueryContext context,
IFullPath fullPath,
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index de1a5a09b7d..6724d196443 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -544,19 +544,33 @@ public class TsFileProcessor {
walNode.onMemTableCreated(workMemTable, tsFileResource.getTsFilePath());
}
+ private static List<List<Pair<IDeviceID, Integer>>> splitTabletByDevice(
+ InsertTabletNode insertTabletNode, List<int[]> rangeList) {
+ List<List<Pair<IDeviceID, Integer>>> deviceEndOffsetPairsList =
+ new ArrayList<>(rangeList.size());
+ for (int[] range : rangeList) {
+ deviceEndOffsetPairsList.add(insertTabletNode.splitByDevice(range[0],
range[1]));
+ }
+ return deviceEndOffsetPairsList;
+ }
+
private long[] scheduleMemoryBlock(
InsertTabletNode insertTabletNode,
List<int[]> rangeList,
+ List<List<Pair<IDeviceID, Integer>>> deviceEndOffsetPairsList,
TSStatus[] results,
long[] infoForMetrics)
throws WriteProcessException {
long memControlStartTime = System.nanoTime();
long[] totalMemIncrements = new long[NUM_MEM_TO_ESTIMATE];
try {
- for (int[] range : rangeList) {
+ for (int rangeIndex = 0; rangeIndex < rangeList.size(); ++rangeIndex) {
+ int[] range = rangeList.get(rangeIndex);
int start = range[0];
int end = range[1];
- long[] memIncrements = checkMemCost(insertTabletNode, start, end,
results);
+ long[] memIncrements =
+ checkMemCost(
+ insertTabletNode, start, end,
deviceEndOffsetPairsList.get(rangeIndex), results);
for (int i = 0; i < memIncrements.length; i++) {
totalMemIncrements[i] += memIncrements[i];
}
@@ -578,11 +592,15 @@ public class TsFileProcessor {
}
private long[] checkMemCost(
- InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results)
+ InsertTabletNode insertTabletNode,
+ int start,
+ int end,
+ List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs,
+ TSStatus[] results)
throws WriteProcessException {
long[] memIncrements;
if (insertTabletNode.isAligned()) {
- memIncrements = checkAlignedMemCost(insertTabletNode, start, end,
results);
+ memIncrements = checkAlignedMemCost(insertTabletNode, start,
deviceEndOffsetPairs, results);
} else {
memIncrements =
checkMemCostAndAddToTspInfoForTablet(
@@ -598,9 +616,11 @@ public class TsFileProcessor {
}
private long[] checkAlignedMemCost(
- InsertTabletNode insertTabletNode, int start, int end, TSStatus[]
results)
+ InsertTabletNode insertTabletNode,
+ int start,
+ List<Pair<IDeviceID, Integer>> deviceEndPosList,
+ TSStatus[] results)
throws WriteProcessException {
- List<Pair<IDeviceID, Integer>> deviceEndPosList =
insertTabletNode.splitByDevice(start, end);
long[] memIncrements = new long[NUM_MEM_TO_ESTIMATE];
int splitStart = start;
for (Pair<IDeviceID, Integer> iDeviceIDIntegerPair : deviceEndPosList) {
@@ -650,11 +670,15 @@ public class TsFileProcessor {
ensureMemTable(infoForMetrics);
workMemTable.checkDataType(insertTabletNode);
+ List<List<Pair<IDeviceID, Integer>>> deviceEndOffsetPairsList =
+ splitTabletByDevice(insertTabletNode, rangeList);
AlignedTVListRamCostSnapshot alignedRamCostSnapshot =
- takeAlignedTVListRamCostSnapshot(workMemTable, insertTabletNode,
rangeList);
+ takeAlignedTVListRamCostSnapshotFromDeviceRanges(
+ workMemTable, insertTabletNode, deviceEndOffsetPairsList);
long[] memIncrements =
- scheduleMemoryBlock(insertTabletNode, rangeList, results,
infoForMetrics);
+ scheduleMemoryBlock(
+ insertTabletNode, rangeList, deviceEndOffsetPairsList, results,
infoForMetrics);
clearDataRegionReplicaSet(insertTabletNode);
long startTime = System.nanoTime();
@@ -708,7 +732,11 @@ public class TsFileProcessor {
if (insertTabletNode.isAligned()) {
pointInserted +=
workMemTable.insertAlignedTablet(
- insertTabletNode, start, end, noFailure ? null : results);
+ insertTabletNode,
+ start,
+ end,
+ noFailure ? null : results,
+ deviceEndOffsetPairsList.get(rangeIndex));
} else {
pointInserted += workMemTable.insertTablet(insertTabletNode,
start, end);
}
@@ -732,7 +760,7 @@ public class TsFileProcessor {
}
final List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs =
- insertTabletNode.splitByDevice(start, end);
+ deviceEndOffsetPairsList.get(rangeIndex);
tsFileResource.updateStartTime(
deviceEndOffsetPairs.get(0).left,
insertTabletNode.getTimes()[start]);
if (!sequence) {
@@ -1409,6 +1437,38 @@ public class TsFileProcessor {
: new AlignedTVListRamCostSnapshot(memTable, alignedDeviceIds);
}
+ private static AlignedTVListRamCostSnapshot
takeAlignedTVListRamCostSnapshotFromDeviceRanges(
+ IMemTable memTable,
+ InsertTabletNode insertTabletNode,
+ List<List<Pair<IDeviceID, Integer>>> deviceEndOffsetPairsList) {
+ if (!insertTabletNode.isAligned() || deviceEndOffsetPairsList.isEmpty()) {
+ return null;
+ }
+
+ IDeviceID singleDeviceId = null;
+ Set<IDeviceID> alignedDeviceIds = null;
+ for (List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs :
deviceEndOffsetPairsList) {
+ for (Pair<IDeviceID, Integer> deviceEndOffsetPair :
deviceEndOffsetPairs) {
+ IDeviceID deviceId = deviceEndOffsetPair.getLeft();
+ if (singleDeviceId == null) {
+ singleDeviceId = deviceId;
+ } else if (!singleDeviceId.equals(deviceId)) {
+ if (alignedDeviceIds == null) {
+ alignedDeviceIds = new HashSet<>();
+ alignedDeviceIds.add(singleDeviceId);
+ }
+ alignedDeviceIds.add(deviceId);
+ }
+ }
+ }
+ if (alignedDeviceIds != null) {
+ return new AlignedTVListRamCostSnapshot(memTable, alignedDeviceIds);
+ }
+ return singleDeviceId == null
+ ? null
+ : new AlignedTVListRamCostSnapshot(memTable, singleDeviceId);
+ }
+
static final class AlignedTVListRamCostSnapshot {
private final IMemTable memTable;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TabletDeviceRangeReusePerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TabletDeviceRangeReusePerformanceTest.java
new file mode 100644
index 00000000000..ebc004d9719
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TabletDeviceRangeReusePerformanceTest.java
@@ -0,0 +1,283 @@
+/*
+ * 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.storageengine.dataregion.memtable;
+
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement;
+import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.Pair;
+import org.junit.Assert;
+import org.junit.Assume;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class TabletDeviceRangeReusePerformanceTest {
+
+ private static final String ENABLED_PROPERTY =
"iotdb.tablet.device.ranges.perf.enabled";
+ private static final String ROWS_PROPERTY =
"iotdb.tablet.device.ranges.perf.rows";
+ private static final String DEVICES_PROPERTY =
"iotdb.tablet.device.ranges.perf.devices";
+ private static final String WARMUP_ITERATIONS_PROPERTY =
+ "iotdb.tablet.device.ranges.perf.warmup.iterations";
+ private static final String ITERATIONS_PROPERTY =
"iotdb.tablet.device.ranges.perf.iterations";
+ private static final String ROUNDS_PROPERTY =
"iotdb.tablet.device.ranges.perf.rounds";
+ private static volatile int benchmarkBlackhole;
+
+ @Test
+ public void tabletDeviceRangeReuseBenchmark() throws IllegalPathException {
+ Assume.assumeTrue(
+ String.format(
+ "Manual performance UT. Enable with -D%s=true, optionally tune
-D%s, -D%s, -D%s, -D%s and -D%s.",
+ ENABLED_PROPERTY,
+ ROWS_PROPERTY,
+ DEVICES_PROPERTY,
+ WARMUP_ITERATIONS_PROPERTY,
+ ITERATIONS_PROPERTY,
+ ROUNDS_PROPERTY),
+ Boolean.getBoolean(ENABLED_PROPERTY));
+ Assume.assumeTrue(
+ "Current-thread CPU time and allocation metrics are required.",
+ ManualPerformanceTestUtils.enableThreadMetrics());
+
+ final int rowCount = Integer.getInteger(ROWS_PROPERTY, 4096);
+ final int deviceCount = Integer.getInteger(DEVICES_PROPERTY, 64);
+ final int warmupIterations =
Integer.getInteger(WARMUP_ITERATIONS_PROPERTY, 1000);
+ final int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 10_000);
+ final int rounds = Integer.getInteger(ROUNDS_PROPERTY, 7);
+ Assert.assertTrue(rowCount > 0);
+ Assert.assertTrue(deviceCount > 0);
+ Assert.assertTrue(rowCount >= deviceCount);
+ Assert.assertTrue(warmupIterations > 0);
+ Assert.assertTrue(iterations > 0);
+ Assert.assertTrue(rounds > 0);
+
+ final BenchmarkNode node = new BenchmarkNode(rowCount, deviceCount);
+ final List<int[]> ranges = Collections.singletonList(new int[] {0,
rowCount});
+ runLegacy(node, ranges, warmupIterations);
+ runCached(node, ranges, warmupIterations);
+ node.resetCalls();
+
+ final Measurement[] legacyMeasurements = new Measurement[rounds];
+ final Measurement[] cachedMeasurements = new Measurement[rounds];
+ for (int round = 0; round < rounds; ++round) {
+ if ((round & 1) == 0) {
+ legacyMeasurements[round] = measureLegacy(node, ranges, iterations);
+ cachedMeasurements[round] = measureCached(node, ranges, iterations);
+ } else {
+ cachedMeasurements[round] = measureCached(node, ranges, iterations);
+ legacyMeasurements[round] = measureLegacy(node, ranges, iterations);
+ }
+ }
+
+ final Summary legacySummary =
+ ManualPerformanceTestUtils.summarize(legacyMeasurements, iterations);
+ final Summary cachedSummary =
+ ManualPerformanceTestUtils.summarize(cachedMeasurements, iterations);
+ final long expectedLegacySplits = (long) iterations * rounds * 4;
+ final long expectedCachedSplits = iterations * rounds;
+ Assert.assertEquals(expectedLegacySplits, node.legacySplitCalls.get());
+ Assert.assertEquals(expectedCachedSplits, node.cachedSplitCalls.get());
+ Assert.assertEquals(node.cachedDeviceIdCalls.get() * 4,
node.legacyDeviceIdCalls.get());
+ printResult(
+ rowCount,
+ deviceCount,
+ warmupIterations,
+ iterations,
+ rounds,
+ legacySummary,
+ cachedSummary,
+ node.legacySplitCalls.get(),
+ node.cachedSplitCalls.get(),
+ node.legacyDeviceIdCalls.get(),
+ node.cachedDeviceIdCalls.get());
+ }
+
+ private static Measurement measureLegacy(BenchmarkNode node, List<int[]>
ranges, int iterations) {
+ return ManualPerformanceTestUtils.measure(iterations, () ->
runLegacyOnce(node, ranges));
+ }
+
+ private static Measurement measureCached(BenchmarkNode node, List<int[]>
ranges, int iterations) {
+ return ManualPerformanceTestUtils.measure(iterations, () ->
runCachedOnce(node, ranges));
+ }
+
+ private static void runLegacy(BenchmarkNode node, List<int[]> ranges, int
iterations) {
+ for (int i = 0; i < iterations; ++i) {
+ runLegacyOnce(node, ranges);
+ }
+ }
+
+ private static void runCached(BenchmarkNode node, List<int[]> ranges, int
iterations) {
+ for (int i = 0; i < iterations; ++i) {
+ runCachedOnce(node, ranges);
+ }
+ }
+
+ private static void runLegacyOnce(BenchmarkNode node, List<int[]> ranges) {
+ long checksum = 0;
+ for (int[] range : ranges) {
+ List<Pair<IDeviceID, Integer>> ramSnapshotRanges =
+ node.legacySplitByDevice(range[0], range[1]);
+ checksum += ramSnapshotRanges.size();
+ }
+ for (int[] range : ranges) {
+ List<Pair<IDeviceID, Integer>> memoryCheckRanges =
+ node.legacySplitByDevice(range[0], range[1]);
+ checksum += memoryCheckRanges.size();
+ }
+ for (int[] range : ranges) {
+ List<Pair<IDeviceID, Integer>> memTableWriteRanges =
+ node.legacySplitByDevice(range[0], range[1]);
+ checksum += memTableWriteRanges.size();
+ }
+ for (int[] range : ranges) {
+ List<Pair<IDeviceID, Integer>> resourceUpdateRanges =
+ node.legacySplitByDevice(range[0], range[1]);
+ checksum += resourceUpdateRanges.size();
+ }
+ benchmarkBlackhole = (int) checksum;
+ }
+
+ private static void runCachedOnce(BenchmarkNode node, List<int[]> ranges) {
+ List<List<Pair<IDeviceID, Integer>>> deviceRanges = new
ArrayList<>(ranges.size());
+ for (int[] range : ranges) {
+ deviceRanges.add(node.cachedSplitByDevice(range[0], range[1]));
+ }
+ long checksum = 0;
+ for (List<Pair<IDeviceID, Integer>> deviceEndOffsetPairs : deviceRanges) {
+ checksum += deviceEndOffsetPairs.size() * 4L;
+ }
+ benchmarkBlackhole = (int) checksum;
+ }
+
+ private static void printResult(
+ int rowCount,
+ int deviceCount,
+ int warmupIterations,
+ int iterations,
+ int rounds,
+ Summary legacySummary,
+ Summary cachedSummary,
+ long legacySplitCalls,
+ long cachedSplitCalls,
+ long legacyDeviceIdCalls,
+ long cachedDeviceIdCalls) {
+ System.out.printf(
+ Locale.ROOT,
+ "Tablet device range reuse benchmark: rows=%d, devices=%d, warmup=%d,
iterations=%d, rounds=%d%n"
+ + "legacy repeated split: cpu=%.2f ns/op, allocation=%.2f
bytes/op, splitCalls=%d, getDeviceIDCalls=%d%n"
+ + "cached ranges: cpu=%.2f ns/op, allocation=%.2f bytes/op,
splitCalls=%d, getDeviceIDCalls=%d%n"
+ + "reduction: cpu=%.2f%%, allocation=%.2f%%%n",
+ rowCount,
+ deviceCount,
+ warmupIterations,
+ iterations,
+ rounds,
+ legacySummary.getCpuNanosPerOperation(),
+ legacySummary.getAllocatedBytesPerOperation(),
+ legacySplitCalls,
+ legacyDeviceIdCalls,
+ cachedSummary.getCpuNanosPerOperation(),
+ cachedSummary.getAllocatedBytesPerOperation(),
+ cachedSplitCalls,
+ cachedDeviceIdCalls,
+ reduction(legacySummary.getCpuNanosPerOperation(),
cachedSummary.getCpuNanosPerOperation()),
+ reduction(
+ legacySummary.getAllocatedBytesPerOperation(),
+ cachedSummary.getAllocatedBytesPerOperation()));
+ }
+
+ private static double reduction(double before, double after) {
+ return before == 0 ? 0 : (before - after) * 100 / before;
+ }
+
+ private static final class BenchmarkNode extends RelationalInsertTabletNode {
+
+ private final IDeviceID[] deviceIds;
+ private final int rowsPerDevice;
+ private final AtomicLong legacySplitCalls = new AtomicLong();
+ private final AtomicLong cachedSplitCalls = new AtomicLong();
+ private final AtomicLong legacyDeviceIdCalls = new AtomicLong();
+ private final AtomicLong cachedDeviceIdCalls = new AtomicLong();
+
+ private BenchmarkNode(int rowCount, int deviceCount) throws
IllegalPathException {
+ super(new PlanNodeId("tablet-device-range-benchmark"));
+ deviceIds = new IDeviceID[deviceCount];
+ for (int i = 0; i < deviceCount; ++i) {
+ deviceIds[i] =
IDeviceID.Factory.DEFAULT_FACTORY.create("root.benchmark.d" + i);
+ }
+ rowsPerDevice = Math.max(1, rowCount / deviceCount);
+ setAligned(true);
+ }
+
+ @Override
+ public IDeviceID getDeviceID(int rowIdx) {
+ return deviceIds[Math.min(rowIdx / rowsPerDevice, deviceIds.length - 1)];
+ }
+
+ @Override
+ public List<Pair<IDeviceID, Integer>> splitByDevice(int start, int end) {
+ return split(start, end, legacySplitCalls, legacyDeviceIdCalls);
+ }
+
+ private List<Pair<IDeviceID, Integer>> legacySplitByDevice(int start, int
end) {
+ return split(start, end, legacySplitCalls, legacyDeviceIdCalls);
+ }
+
+ private List<Pair<IDeviceID, Integer>> cachedSplitByDevice(int start, int
end) {
+ return split(start, end, cachedSplitCalls, cachedDeviceIdCalls);
+ }
+
+ private List<Pair<IDeviceID, Integer>> split(
+ int start, int end, AtomicLong splitCounter, AtomicLong
deviceIdCounter) {
+ splitCounter.incrementAndGet();
+ final List<Pair<IDeviceID, Integer>> result = new ArrayList<>();
+ IDeviceID previous = getDeviceID(start, deviceIdCounter);
+ for (int i = start + 1; i < end; ++i) {
+ IDeviceID current = getDeviceID(i, deviceIdCounter);
+ if (!current.equals(previous)) {
+ result.add(new Pair<>(previous, i));
+ previous = getDeviceID(i, deviceIdCounter);
+ }
+ }
+ result.add(new Pair<>(previous, end));
+ return result;
+ }
+
+ private IDeviceID getDeviceID(int rowIdx, AtomicLong counter) {
+ counter.incrementAndGet();
+ return deviceIds[Math.min(rowIdx / rowsPerDevice, deviceIds.length - 1)];
+ }
+
+ private void resetCalls() {
+ legacySplitCalls.set(0);
+ cachedSplitCalls.set(0);
+ legacyDeviceIdCalls.set(0);
+ cachedDeviceIdCalls.set(0);
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
index e1b402a64dd..f1ca1d597c3 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java
@@ -86,6 +86,7 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
import static junit.framework.TestCase.assertTrue;
import static
org.apache.iotdb.db.storageengine.dataregion.DataRegionTest.buildInsertRowNodeByTSRecord;
@@ -762,6 +763,25 @@ public class TsFileProcessorTest {
Assert.assertEquals(1, relationalSplitCalls[0]);
}
+ @Test
+ public void alignedTabletReusesDeviceRangesAcrossWriteStages()
+ throws IOException, WriteProcessException, IllegalPathException {
+ processor = newTestProcessor(filePath + ".device-range-reuse");
+ final AtomicInteger splitCalls = new AtomicInteger();
+ final int rowCount = 4;
+ final List<int[]> rangeList =
+ Arrays.asList(new int[] {0, rowCount / 2}, new int[] {rowCount / 2,
rowCount});
+
+ processor.insertTablet(
+ genCountingAlignedTablet(new String[] {"s0"}, rowCount, 0, splitCalls),
+ rangeList,
+ new TSStatus[rowCount],
+ true,
+ new long[5]);
+
+ Assert.assertEquals(rangeList.size(), splitCalls.get());
+ }
+
@Test
public void alignedTvListRamCostTest2()
throws MetadataException, WriteProcessException, IOException {
@@ -1596,6 +1616,43 @@ public class TsFileProcessorTest {
rowCount);
}
+ private InsertTabletNode genCountingAlignedTablet(
+ String[] measurements, int rowCount, long startTime, AtomicInteger
splitCalls)
+ throws IllegalPathException {
+ TSDataType[] dataTypes = new TSDataType[measurements.length];
+ MeasurementSchema[] schemas = new MeasurementSchema[measurements.length];
+ Object[] columns = new Object[measurements.length];
+ for (int i = 0; i < measurements.length; i++) {
+ dataTypes[i] = TSDataType.INT32;
+ schemas[i] = new MeasurementSchema(measurements[i], TSDataType.INT32,
encoding);
+ columns[i] = new int[rowCount];
+ }
+ long[] times = new long[rowCount];
+ for (int row = 0; row < rowCount; row++) {
+ times[row] = startTime + row;
+ for (Object column : columns) {
+ ((int[]) column)[row] = row;
+ }
+ }
+ return new InsertTabletNode(
+ new QueryId("test_write").genPlanNodeId(),
+ new PartialPath(deviceId),
+ true,
+ measurements,
+ dataTypes,
+ schemas,
+ times,
+ null,
+ columns,
+ rowCount) {
+ @Override
+ public List<Pair<IDeviceID, Integer>> splitByDevice(int start, int end) {
+ splitCalls.incrementAndGet();
+ return super.splitByDevice(start, end);
+ }
+ };
+ }
+
private InsertTabletNode genInsertTableNode(long startTime, boolean
isAligned)
throws IllegalPathException {
String deviceId = "root.sg.device5";
