Skip to content

Commit 44d4f6d

Browse files
CaideyipiCaideyipi
authored
Optimized write performace by reducing separators (#17670)
1 parent a565a06 commit 44d4f6d

13 files changed

Lines changed: 463 additions & 33 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -76,7 +76,6 @@ public TSStatus visitRelationalInsertRows(RelationalInsertRowsNode node, DataReg
7676
public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
7777
try {
7878
dataRegion.insert(node);
79-
dataRegion.insertSeparatorToWAL();
8079
return StatusUtils.OK;
8180
} catch (OutOfTTLException e) {
8281
LOGGER.warn(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage());
@@ -100,7 +99,6 @@ public TSStatus visitRelationalInsertTablet(
10099
public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion dataRegion) {
101100
try {
102101
dataRegion.insertTablet(node);
103-
dataRegion.insertSeparatorToWAL();
104102
return StatusUtils.OK;
105103
} catch (final OutOfTTLException e) {
106104
LOGGER.debug(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage());
@@ -137,7 +135,6 @@ public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion
137135
public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
138136
try {
139137
dataRegion.insert(node);
140-
dataRegion.insertSeparatorToWAL();
141138
return StatusUtils.OK;
142139
} catch (WriteProcessRejectException e) {
143140
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());
@@ -174,7 +171,6 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
174171
public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion dataRegion) {
175172
try {
176173
dataRegion.insertTablets(node);
177-
dataRegion.insertSeparatorToWAL();
178174
return StatusUtils.OK;
179175
} catch (WriteProcessRejectException e) {
180176
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());
@@ -209,7 +205,6 @@ public TSStatus visitInsertRowsOfOneDevice(
209205
InsertRowsOfOneDeviceNode node, DataRegion dataRegion) {
210206
try {
211207
dataRegion.insert(node);
212-
dataRegion.insertSeparatorToWAL();
213208
return StatusUtils.OK;
214209
} catch (WriteProcessRejectException e) {
215210
LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage());

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowNode.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -621,7 +621,7 @@ private int serializeMeasurementsAndValuesSize() {
621621
@Override
622622
public void serializeToWAL(IWALByteBufferView buffer) {
623623
buffer.putShort(getType().getNodeType());
624-
buffer.putLong(searchIndex);
624+
buffer.putLong(getEncodedSearchIndex());
625625
subSerialize(buffer);
626626
}
627627

@@ -702,7 +702,7 @@ private void putDataTypesAndValues(IWALByteBufferView buffer) {
702702
public static InsertRowNode deserializeFromWAL(DataInputStream stream) throws IOException {
703703
long searchIndex = stream.readLong();
704704
InsertRowNode insertNode = subDeserializeFromWAL(stream);
705-
insertNode.setSearchIndex(searchIndex);
705+
insertNode.setSearchIndexFromWAL(searchIndex);
706706
return insertNode;
707707
}
708708

@@ -793,7 +793,7 @@ public void fillDataTypesAndValuesFromWAL(DataInputStream stream) throws IOExcep
793793
public static InsertRowNode deserializeFromWAL(ByteBuffer buffer) {
794794
long searchIndex = buffer.getLong();
795795
InsertRowNode insertNode = subDeserializeFromWAL(buffer);
796-
insertNode.setSearchIndex(searchIndex);
796+
insertNode.setSearchIndexFromWAL(searchIndex);
797797
return insertNode;
798798
}
799799

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertRowsNode.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -350,7 +350,7 @@ private int subSerializeSize() {
350350
@Override
351351
public void serializeToWAL(IWALByteBufferView buffer) {
352352
buffer.putShort(getType().getNodeType());
353-
buffer.putLong(searchIndex);
353+
buffer.putLong(getEncodedSearchIndex());
354354
subSerialize(buffer);
355355
}
356356

@@ -378,7 +378,7 @@ public static InsertRowsNode deserializeFromWAL(DataInputStream stream) throws I
378378
InsertRowNode insertRowNode = InsertRowNode.subDeserializeFromWAL(stream);
379379
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
380380
}
381-
insertRowsNode.setSearchIndex(searchIndex);
381+
insertRowsNode.setSearchIndexFromWAL(searchIndex);
382382
return insertRowsNode;
383383
}
384384

@@ -398,7 +398,7 @@ public static InsertRowsNode deserializeFromWAL(ByteBuffer buffer) {
398398
InsertRowNode insertRowNode = InsertRowNode.subDeserializeFromWAL(buffer);
399399
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
400400
}
401-
insertRowsNode.setSearchIndex(searchIndex);
401+
insertRowsNode.setSearchIndexFromWAL(searchIndex);
402402
return insertRowsNode;
403403
}
404404

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/InsertTabletNode.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -905,7 +905,7 @@ public void serializeToWAL(IWALByteBufferView buffer, List<int[]> rangeList) {
905905
}
906906

907907
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
908-
buffer.putLong(searchIndex);
908+
buffer.putLong(getEncodedSearchIndex());
909909
WALWriteUtils.write(targetPath.getFullPath(), buffer);
910910
// data types are serialized in measurement schemas
911911
writeMeasurementSchemas(buffer);
@@ -1037,7 +1037,7 @@ public static InsertTabletNode deserializeFromWAL(DataInputStream stream) throws
10371037
}
10381038

10391039
protected void subDeserializeFromWAL(DataInputStream stream) throws IOException {
1040-
searchIndex = stream.readLong();
1040+
setSearchIndexFromWAL(stream.readLong());
10411041
try {
10421042
targetPath = readTargetPath(stream);
10431043
} catch (IllegalPathException e) {
@@ -1073,7 +1073,7 @@ public static InsertTabletNode deserializeFromWAL(ByteBuffer buffer) {
10731073
}
10741074

10751075
protected void subDeserializeFromWAL(ByteBuffer buffer) {
1076-
searchIndex = buffer.getLong();
1076+
setSearchIndexFromWAL(buffer.getLong());
10771077
try {
10781078
targetPath = readTargetPath(buffer);
10791079
} catch (IllegalPathException e) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowNode.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -128,14 +128,14 @@ public static RelationalInsertRowNode deserializeFromWAL(DataInputStream stream)
128128
throws IOException {
129129
long searchIndex = stream.readLong();
130130
RelationalInsertRowNode insertNode = subDeserializeFromWAL(stream);
131-
insertNode.setSearchIndex(searchIndex);
131+
insertNode.setSearchIndexFromWAL(searchIndex);
132132
return insertNode;
133133
}
134134

135135
public static RelationalInsertRowNode deserializeFromWAL(ByteBuffer buffer) {
136136
long searchIndex = buffer.getLong();
137137
RelationalInsertRowNode insertNode = subDeserializeFromWAL(buffer);
138-
insertNode.setSearchIndex(searchIndex);
138+
insertNode.setSearchIndexFromWAL(searchIndex);
139139
return insertNode;
140140
}
141141

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertRowsNode.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -124,7 +124,7 @@ public static RelationalInsertRowsNode deserializeFromWAL(DataInputStream stream
124124
RelationalInsertRowNode insertRowNode = RelationalInsertRowNode.subDeserializeFromWAL(stream);
125125
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
126126
}
127-
insertRowsNode.setSearchIndex(searchIndex);
127+
insertRowsNode.setSearchIndexFromWAL(searchIndex);
128128
return insertRowsNode;
129129
}
130130

@@ -144,7 +144,7 @@ public static RelationalInsertRowsNode deserializeFromWAL(ByteBuffer buffer) {
144144
RelationalInsertRowNode insertRowNode = RelationalInsertRowNode.subDeserializeFromWAL(buffer);
145145
insertRowsNode.addOneInsertRowNode(insertRowNode, i);
146146
}
147-
insertRowsNode.setSearchIndex(searchIndex);
147+
insertRowsNode.setSearchIndexFromWAL(searchIndex);
148148
return insertRowsNode;
149149
}
150150

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/SearchNode.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,15 +28,22 @@
2828

2929
public abstract class SearchNode extends WritePlanNode implements ComparableConsensusRequest {
3030

31+
private static final long LAST_FRAGMENT_MASK = Long.MIN_VALUE;
32+
3133
/** this insert node doesn't need to participate in iot consensus */
3234
public static final long NO_CONSENSUS_INDEX = ConsensusReqReader.DEFAULT_SEARCH_INDEX;
3335

36+
// Preserve last-fragment state for WAL entries that do not have a consensus search index.
37+
private static final long NO_CONSENSUS_INDEX_WITH_LAST_FRAGMENT = Long.MIN_VALUE;
38+
3439
/**
3540
* this index is used by wal search, its order should be protected by the upper layer, and the
3641
* value should start from 1
3742
*/
3843
protected long searchIndex = NO_CONSENSUS_INDEX;
3944

45+
protected boolean isLastFragment = false;
46+
4047
protected SearchNode(PlanNodeId id) {
4148
super(id);
4249
}
@@ -51,5 +58,43 @@ public SearchNode setSearchIndex(long searchIndex) {
5158
return this;
5259
}
5360

61+
public boolean isLastFragment() {
62+
return isLastFragment;
63+
}
64+
65+
public SearchNode setLastFragment(boolean lastFragment) {
66+
isLastFragment = lastFragment;
67+
return this;
68+
}
69+
70+
protected long getEncodedSearchIndex() {
71+
if (!isLastFragment) {
72+
return searchIndex;
73+
}
74+
if (searchIndex == NO_CONSENSUS_INDEX) {
75+
return NO_CONSENSUS_INDEX_WITH_LAST_FRAGMENT;
76+
}
77+
return searchIndex | LAST_FRAGMENT_MASK;
78+
}
79+
80+
public static long extractSearchIndex(long encodedSearchIndex) {
81+
if (encodedSearchIndex == NO_CONSENSUS_INDEX
82+
|| encodedSearchIndex == NO_CONSENSUS_INDEX_WITH_LAST_FRAGMENT) {
83+
return NO_CONSENSUS_INDEX;
84+
}
85+
return encodedSearchIndex & ~LAST_FRAGMENT_MASK;
86+
}
87+
88+
public static boolean isLastFragment(long encodedSearchIndex) {
89+
return encodedSearchIndex == NO_CONSENSUS_INDEX_WITH_LAST_FRAGMENT
90+
|| (encodedSearchIndex != NO_CONSENSUS_INDEX
91+
&& (encodedSearchIndex & LAST_FRAGMENT_MASK) != 0);
92+
}
93+
94+
protected void setSearchIndexFromWAL(long encodedSearchIndex) {
95+
this.searchIndex = extractSearchIndex(encodedSearchIndex);
96+
this.isLastFragment = isLastFragment(encodedSearchIndex);
97+
}
98+
5499
public abstract SearchNode merge(List<SearchNode> searchNodes);
55100
}

0 commit comments

Comments
 (0)