Skip to content

Commit 011bdec

Browse files
CaideyipiCaideyipi
authored
Fix WAL search index freezing (#17811)
1 parent 99f0af1 commit 011bdec

11 files changed

Lines changed: 149 additions & 33 deletions

File tree

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -122,7 +122,7 @@ public static DeleteDataNode deserializeFromWAL(DataInputStream stream) throws I
122122

123123
DeleteDataNode deleteDataNode =
124124
new DeleteDataNode(new PlanNodeId(""), pathList, deleteStartTime, deleteEndTime);
125-
deleteDataNode.setSearchIndex(searchIndex);
125+
deleteDataNode.setSearchIndexFromWAL(searchIndex);
126126
return deleteDataNode;
127127
}
128128

@@ -143,7 +143,7 @@ public static DeleteDataNode deserializeFromWAL(ByteBuffer buffer) {
143143

144144
DeleteDataNode deleteDataNode =
145145
new DeleteDataNode(new PlanNodeId(""), pathList, deleteStartTime, deleteEndTime);
146-
deleteDataNode.setSearchIndex(searchIndex);
146+
deleteDataNode.setSearchIndexFromWAL(searchIndex);
147147
return deleteDataNode;
148148
}
149149

@@ -233,8 +233,12 @@ public int serializedSize() {
233233

234234
@Override
235235
public void serializeToWAL(IWALByteBufferView buffer) {
236+
serializeToWAL(buffer, getEncodedSearchIndex());
237+
}
238+
239+
public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) {
236240
buffer.putShort(PlanNodeType.DELETE_DATA.getNodeType());
237-
buffer.putLong(searchIndex);
241+
buffer.putLong(encodedSearchIndex);
238242
buffer.putInt(pathList.size());
239243
for (PartialPath path : pathList) {
240244
WALWriteUtils.write(path.getFullPath(), buffer);

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -620,8 +620,12 @@ private int serializeMeasurementsAndValuesSize() {
620620
*/
621621
@Override
622622
public void serializeToWAL(IWALByteBufferView buffer) {
623+
serializeToWAL(buffer, getEncodedSearchIndex());
624+
}
625+
626+
public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) {
623627
buffer.putShort(getType().getNodeType());
624-
buffer.putLong(getEncodedSearchIndex());
628+
buffer.putLong(encodedSearchIndex);
625629
subSerialize(buffer);
626630
}
627631

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -349,8 +349,12 @@ private int subSerializeSize() {
349349
*/
350350
@Override
351351
public void serializeToWAL(IWALByteBufferView buffer) {
352+
serializeToWAL(buffer, getEncodedSearchIndex());
353+
}
354+
355+
public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) {
352356
buffer.putShort(getType().getNodeType());
353-
buffer.putLong(getEncodedSearchIndex());
357+
buffer.putLong(encodedSearchIndex);
354358
subSerialize(buffer);
355359
}
356360

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

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -900,12 +900,21 @@ public void serializeToWAL(IWALByteBufferView buffer) {
900900
}
901901

902902
public void serializeToWAL(IWALByteBufferView buffer, List<int[]> rangeList) {
903+
serializeToWAL(buffer, rangeList, getEncodedSearchIndex());
904+
}
905+
906+
public void serializeToWAL(
907+
IWALByteBufferView buffer, List<int[]> rangeList, long encodedSearchIndex) {
903908
buffer.putShort(getType().getNodeType());
904-
subSerialize(buffer, rangeList);
909+
subSerialize(buffer, rangeList, encodedSearchIndex);
905910
}
906911

907912
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
908-
buffer.putLong(getEncodedSearchIndex());
913+
subSerialize(buffer, rangeList, getEncodedSearchIndex());
914+
}
915+
916+
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList, long encodedSearchIndex) {
917+
buffer.putLong(encodedSearchIndex);
909918
WALWriteUtils.write(targetPath.getFullPath(), buffer);
910919
// data types are serialized in measurement schemas
911920
writeMeasurementSchemas(buffer);

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -109,8 +109,12 @@ public String getFilePathString() {
109109

110110
@Override
111111
public void serializeToWAL(IWALByteBufferView buffer) {
112+
serializeToWAL(buffer, getEncodedSearchIndex());
113+
}
114+
115+
public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) {
112116
buffer.putShort(getType().getNodeType());
113-
buffer.putLong(searchIndex);
117+
buffer.putLong(encodedSearchIndex);
114118
buffer.put((byte) (isEOF ? 1 : 0));
115119
buffer.putLong(offset);
116120
try {
@@ -138,7 +142,7 @@ public static ObjectNode deserializeFromWAL(DataInputStream stream) throws IOExc
138142
IObjectPath filePath = IObjectPath.getDeserializer().deserializeFrom(stream);
139143
int contentLength = stream.readInt();
140144
ObjectNode objectNode = new ObjectNode(isEOF, offset, contentLength, filePath);
141-
objectNode.setSearchIndex(searchIndex);
145+
objectNode.setSearchIndexFromWAL(searchIndex);
142146
return objectNode;
143147
}
144148

@@ -163,7 +167,7 @@ public static ObjectNode deserializeFromWAL(ByteBuffer buffer) {
163167
}
164168

165169
ObjectNode objectNode = new ObjectNode(isEOF, offset, contents, filePath);
166-
objectNode.setSearchIndex(searchIndex);
170+
objectNode.setSearchIndexFromWAL(searchIndex);
167171
return objectNode;
168172
}
169173

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -130,7 +130,7 @@ public static RelationalDeleteDataNode deserializeFromWAL(DataInputStream stream
130130

131131
RelationalDeleteDataNode deleteDataNode =
132132
new RelationalDeleteDataNode(new PlanNodeId(""), modEntries, databaseName);
133-
deleteDataNode.setSearchIndex(searchIndex);
133+
deleteDataNode.setSearchIndexFromWAL(searchIndex);
134134
return deleteDataNode;
135135
}
136136

@@ -145,7 +145,7 @@ public static RelationalDeleteDataNode deserializeFromWAL(ByteBuffer buffer) {
145145

146146
RelationalDeleteDataNode deleteDataNode =
147147
new RelationalDeleteDataNode(new PlanNodeId(""), modEntries, databaseName);
148-
deleteDataNode.setSearchIndex(searchIndex);
148+
deleteDataNode.setSearchIndexFromWAL(searchIndex);
149149
return deleteDataNode;
150150
}
151151

@@ -225,8 +225,12 @@ public int serializedSize() {
225225

226226
@Override
227227
public void serializeToWAL(IWALByteBufferView buffer) {
228+
serializeToWAL(buffer, getEncodedSearchIndex());
229+
}
230+
231+
public void serializeToWAL(IWALByteBufferView buffer, long encodedSearchIndex) {
228232
buffer.putShort(PlanNodeType.RELATIONAL_DELETE_DATA.getNodeType());
229-
buffer.putLong(searchIndex);
233+
buffer.putLong(encodedSearchIndex);
230234
try {
231235
ReadWriteForEncodingUtils.writeVarInt(modEntries.size(), buffer);
232236
for (TableDeletionEntry modEntry : modEntries) {

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

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -278,7 +278,12 @@ public void subDeserialize(ByteBuffer buffer) {
278278

279279
@Override
280280
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList) {
281-
super.subSerialize(buffer, rangeList);
281+
subSerialize(buffer, rangeList, getEncodedSearchIndex());
282+
}
283+
284+
@Override
285+
void subSerialize(IWALByteBufferView buffer, List<int[]> rangeList, long encodedSearchIndex) {
286+
super.subSerialize(buffer, rangeList, encodedSearchIndex);
282287
for (int i = 0; i < measurements.length; i++) {
283288
if (measurements[i] != null) {
284289
buffer.put(columnCategories[i].getCategory());

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

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,11 @@ public SearchNode setLastFragment(boolean lastFragment) {
6767
return this;
6868
}
6969

70-
protected long getEncodedSearchIndex() {
70+
public long getEncodedSearchIndex() {
71+
return encodeSearchIndex(searchIndex, isLastFragment);
72+
}
73+
74+
public static long encodeSearchIndex(long searchIndex, boolean isLastFragment) {
7175
if (!isLastFragment) {
7276
return searchIndex;
7377
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java‎

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -26,10 +26,6 @@
2626
import org.apache.iotdb.db.conf.IoTDBConfig;
2727
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2828
import org.apache.iotdb.db.i18n.StorageEngineMessages;
29-
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
30-
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
31-
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode;
32-
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
3329
import org.apache.iotdb.db.service.metrics.WritingMetrics;
3430
import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.Checkpoint;
3531
import org.apache.iotdb.db.storageengine.dataregion.wal.checkpoint.CheckpointManager;
@@ -338,15 +334,7 @@ private void handleInfoEntry(WALEntry walEntry) {
338334
// parse search index
339335
long searchIndex = DEFAULT_SEARCH_INDEX;
340336
if (walEntry.getType().needSearch()) {
341-
if (walEntry.getType() == WALEntryType.DELETE_DATA_NODE) {
342-
searchIndex = ((DeleteDataNode) walEntry.getValue()).getSearchIndex();
343-
} else if (walEntry.getType() == WALEntryType.RELATIONAL_DELETE_DATA_NODE) {
344-
searchIndex = ((RelationalDeleteDataNode) walEntry.getValue()).getSearchIndex();
345-
} else if (walEntry.getType() == WALEntryType.OBJECT_FILE_NODE) {
346-
searchIndex = ((ObjectNode) walEntry.getValue()).getSearchIndex();
347-
} else {
348-
searchIndex = ((InsertNode) walEntry.getValue()).getSearchIndex();
349-
}
337+
searchIndex = ((WALInfoEntry) walEntry).getSearchIndex();
350338
if (searchIndex != DEFAULT_SEARCH_INDEX) {
351339
currentSearchIndex = searchIndex;
352340
currentFileStatus = WALFileStatus.CONTAINS_SEARCH_INDEX;

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALInfoEntry.java‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,14 @@
2222
import org.apache.iotdb.db.conf.IoTDBConfig;
2323
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2424
import org.apache.iotdb.db.i18n.StorageEngineMessages;
25+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
2526
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
27+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
28+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
2629
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
2730
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.ObjectNode;
31+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
32+
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
2833
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
2934
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode;
3035

@@ -44,9 +49,11 @@ public class WALInfoEntry extends WALEntry {
4449

4550
// extra info for InsertTablet type value
4651
private TabletInfo tabletInfo;
52+
private final Long encodedSearchIndex;
4753

4854
public WALInfoEntry(long memTableId, WALEntryValue value, boolean wait) {
4955
super(memTableId, value, wait);
56+
encodedSearchIndex = freezeEncodedSearchIndex(value);
5057
}
5158

5259
public WALInfoEntry(long memTableId, WALEntryValue value) {
@@ -65,6 +72,7 @@ public WALInfoEntry(long memTableId, InsertTabletNode value, List<int[]> tabletR
6572

6673
WALInfoEntry(WALEntryType type, long memTableId, WALEntryValue value) {
6774
super(type, memTableId, value, false);
75+
encodedSearchIndex = freezeEncodedSearchIndex(value);
6876
if (value instanceof InsertTabletNode) {
6977
tabletInfo =
7078
new TabletInfo(
@@ -83,15 +91,26 @@ public void serialize(IWALByteBufferView buffer) {
8391
buffer.putLong(memTableId);
8492
switch (type) {
8593
case INSERT_TABLET_NODE:
86-
((InsertTabletNode) value).serializeToWAL(buffer, tabletInfo.tabletRangeList);
94+
((InsertTabletNode) value)
95+
.serializeToWAL(buffer, tabletInfo.tabletRangeList, encodedSearchIndex);
8796
break;
8897
case INSERT_ROW_NODE:
98+
((InsertRowNode) value).serializeToWAL(buffer, encodedSearchIndex);
99+
break;
89100
case INSERT_ROWS_NODE:
101+
((InsertRowsNode) value).serializeToWAL(buffer, encodedSearchIndex);
102+
break;
90103
case DELETE_DATA_NODE:
104+
((DeleteDataNode) value).serializeToWAL(buffer, encodedSearchIndex);
105+
break;
91106
case RELATIONAL_DELETE_DATA_NODE:
107+
((RelationalDeleteDataNode) value).serializeToWAL(buffer, encodedSearchIndex);
108+
break;
109+
case OBJECT_FILE_NODE:
110+
((ObjectNode) value).serializeToWAL(buffer, encodedSearchIndex);
111+
break;
92112
case MEMORY_TABLE_SNAPSHOT:
93113
case CONTINUOUS_SAME_SEARCH_INDEX_SEPARATOR_NODE:
94-
case OBJECT_FILE_NODE:
95114
value.serializeToWAL(buffer);
96115
break;
97116
case MEMORY_TABLE_CHECKPOINT:
@@ -101,12 +120,25 @@ public void serialize(IWALByteBufferView buffer) {
101120
}
102121
}
103122

123+
public long getSearchIndex() {
124+
return encodedSearchIndex == null
125+
? SearchNode.NO_CONSENSUS_INDEX
126+
: SearchNode.extractSearchIndex(encodedSearchIndex);
127+
}
128+
129+
private static Long freezeEncodedSearchIndex(WALEntryValue value) {
130+
return value instanceof SearchNode ? ((SearchNode) value).getEncodedSearchIndex() : null;
131+
}
132+
104133
private static class TabletInfo {
105134
// ranges of insert tablet
106135
private final List<int[]> tabletRangeList;
107136

108137
public TabletInfo(List<int[]> tabletRangeList) {
109-
this.tabletRangeList = new ArrayList<>(tabletRangeList);
138+
this.tabletRangeList = new ArrayList<>(tabletRangeList.size());
139+
for (int[] range : tabletRangeList) {
140+
this.tabletRangeList.add(Arrays.copyOf(range, range.length));
141+
}
110142
}
111143

112144
public int getRangeRowCount() {
@@ -178,7 +210,7 @@ public long getMemorySize() {
178210

179211
@Override
180212
public int hashCode() {
181-
return Objects.hash(super.hashCode(), tabletInfo);
213+
return Objects.hash(super.hashCode(), tabletInfo, encodedSearchIndex);
182214
}
183215

184216
@Override
@@ -187,6 +219,7 @@ public boolean equals(Object obj) {
187219
return false;
188220
}
189221
WALInfoEntry other = (WALInfoEntry) obj;
190-
return Objects.equals(this.tabletInfo, other.tabletInfo);
222+
return Objects.equals(this.tabletInfo, other.tabletInfo)
223+
&& Objects.equals(this.encodedSearchIndex, other.encodedSearchIndex);
191224
}
192225
}

0 commit comments

Comments
 (0)