This is an automated email from the ASF dual-hosted git repository.
qiaojialin 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 1ff9604bc6 [IOTDB-4152]Fix tablet split by time partition error (#7066)
1ff9604bc6 is described below
commit 1ff9604bc66143f43e5b833446194c1dd0a482ea
Author: Haonan <[email protected]>
AuthorDate: Mon Aug 22 21:32:37 2022 +0800
[IOTDB-4152]Fix tablet split by time partition error (#7066)
---
.../planner/plan/node/write/InsertTabletNode.java | 46 ++++++++++------------
.../plan/node/write/WritePlanNodeSplitTest.java | 26 +++++++++++-
2 files changed, 45 insertions(+), 27 deletions(-)
diff --git
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
index 1a660edc7b..06512c6a58 100644
---
a/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
+++
b/server/src/main/java/org/apache/iotdb/db/mpp/plan/planner/plan/node/write/InsertTabletNode.java
@@ -236,25 +236,20 @@ public class InsertTabletNode extends InsertNode
implements WALEntryValue {
// generate a new times and values
locs = entry.getValue();
// Avoid using system arraycopy when there is no need to split
- if (splitMap.size() == 1) {
+ if (splitMap.size() == 1 && locs.size() == 2) {
setRange(locs);
setDataRegionReplicaSet(entry.getKey());
result.add(this);
return result;
}
- int count = 0;
- for (int i = 0; i < locs.size(); i += 2) {
- int start = locs.get(i);
- int end = locs.get(i + 1);
- count += end - start;
- }
- long[] subTimes = new long[count];
- int destLoc = 0;
- Object[] values = initTabletValues(dataTypes.length, count, dataTypes);
- BitMap[] bitMaps = this.bitMaps == null ? null :
initBitmaps(dataTypes.length, count);
for (int i = 0; i < locs.size(); i += 2) {
int start = locs.get(i);
int end = locs.get(i + 1);
+ int count = end - start;
+ long[] subTimes = new long[count];
+ int destLoc = 0;
+ Object[] values = initTabletValues(dataTypes.length, count, dataTypes);
+ BitMap[] bitMaps = this.bitMaps == null ? null :
initBitmaps(dataTypes.length, count);
System.arraycopy(times, start, subTimes, destLoc, end - start);
for (int k = 0; k < values.length; k++) {
System.arraycopy(columns[k], start, values[k], destLoc, end - start);
@@ -262,22 +257,21 @@ public class InsertTabletNode extends InsertNode
implements WALEntryValue {
BitMap.copyOfRange(this.bitMaps[k], start, bitMaps[k], destLoc,
end - start);
}
}
- destLoc += end - start;
+ InsertTabletNode subNode =
+ new InsertTabletNode(
+ getPlanNodeId(),
+ devicePath,
+ isAligned,
+ measurements,
+ dataTypes,
+ subTimes,
+ bitMaps,
+ values,
+ subTimes.length);
+ subNode.setRange(locs);
+ subNode.setDataRegionReplicaSet(entry.getKey());
+ result.add(subNode);
}
- InsertTabletNode subNode =
- new InsertTabletNode(
- getPlanNodeId(),
- devicePath,
- isAligned,
- measurements,
- dataTypes,
- subTimes,
- bitMaps,
- values,
- subTimes.length);
- subNode.setRange(locs);
- subNode.setDataRegionReplicaSet(entry.getKey());
- result.add(subNode);
}
return result;
}
diff --git
a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/WritePlanNodeSplitTest.java
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/WritePlanNodeSplitTest.java
index 30e3177b26..c5a5466b2f 100644
---
a/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/WritePlanNodeSplitTest.java
+++
b/server/src/test/java/org/apache/iotdb/db/mpp/plan/plan/node/write/WritePlanNodeSplitTest.java
@@ -90,6 +90,7 @@ public class WritePlanNodeSplitTest {
dataPartitionMap = new HashMap<>();
Map<TSeriesPartitionSlot, Map<TTimePartitionSlot, List<TRegionReplicaSet>>>
seriesPartitionSlotMap = new HashMap<>();
+ // sg1 has 5 data regions
for (int i = 0; i < seriesSlotPartitionNum; i++) {
Map<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotMap =
new HashMap<>();
for (int t = 0; t < 5; t++) {
@@ -105,6 +106,7 @@ public class WritePlanNodeSplitTest {
dataPartitionMap.put("root.sg1", seriesPartitionSlotMap);
+ // sg2 has 1 data region
seriesPartitionSlotMap = new HashMap<>();
for (int i = 0; i < seriesSlotPartitionNum; i++) {
Map<TTimePartitionSlot, List<TRegionReplicaSet>> timePartitionSlotMap =
new HashMap<>();
@@ -113,7 +115,7 @@ public class WritePlanNodeSplitTest {
new TTimePartitionSlot(t *
StorageEngineV2.getTimePartitionInterval()),
Collections.singletonList(
new TRegionReplicaSet(
- new TConsensusGroupId(TConsensusGroupType.DataRegion, 5 -
t), null)));
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 5),
null)));
}
seriesPartitionSlotMap.put(new TSeriesPartitionSlot(i),
timePartitionSlotMap);
@@ -194,6 +196,28 @@ public class WritePlanNodeSplitTest {
for (WritePlanNode insertNode : insertTabletNodeList) {
Assert.assertEquals(((InsertTabletNode) insertNode).getTimes().length,
2);
}
+
+ insertTabletNode = new InsertTabletNode(new PlanNodeId("plan node 2"));
+
+ insertTabletNode.setDevicePath(new PartialPath("root.sg2.d1"));
+ insertTabletNode.setTimes(new long[] {1, 60, 120, 180, 270, 290, 360, 375,
440, 470});
+ insertTabletNode.setDataTypes(new TSDataType[] {TSDataType.INT32});
+ insertTabletNode.setColumns(new Object[] {new int[] {10, 20, 30, 40, 50,
60, 70, 80, 90, 100}});
+
+ dataPartitionQueryParam = new DataPartitionQueryParam();
+
dataPartitionQueryParam.setDevicePath(insertTabletNode.getDevicePath().getFullPath());
+
dataPartitionQueryParam.setTimePartitionSlotList(insertTabletNode.getTimePartitionSlots());
+
+ dataPartition =
getDataPartition(Collections.singletonList(dataPartitionQueryParam));
+ analysis = new Analysis();
+ analysis.setDataPartitionInfo(dataPartition);
+
+ insertTabletNodeList = insertTabletNode.splitByPartition(analysis);
+
+ Assert.assertEquals(5, insertTabletNodeList.size());
+ for (WritePlanNode insertNode : insertTabletNodeList) {
+ Assert.assertEquals(((InsertTabletNode) insertNode).getTimes().length,
2);
+ }
}
@Test
