Skip to content

Commit 767bdfa

Browse files
committed
fix CQ snapshot compatibility
1 parent 7026e36 commit 767bdfa

2 files changed

Lines changed: 59 additions & 4 deletions

File tree

  • iotdb-core/confignode/src

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/cq/CQInfo.java‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -317,13 +317,21 @@ private void serialize(OutputStream stream) throws IOException {
317317
private void deserialize(InputStream stream) throws IOException {
318318
int markerOrSize = ReadWriteIOUtils.readInt(stream);
319319
final boolean structured;
320+
final boolean hasOccurrenceIndex;
320321
final int size;
321-
if (markerOrSize == SNAPSHOT_VERSION || markerOrSize == PREVIOUS_SNAPSHOT_VERSION) {
322+
if (markerOrSize == SNAPSHOT_VERSION) {
322323
structured = true;
324+
hasOccurrenceIndex = true;
325+
size = ReadWriteIOUtils.readInt(stream);
326+
} else if (markerOrSize == PREVIOUS_SNAPSHOT_VERSION) {
327+
// The first structured snapshot format did not persist nextOccurrenceIndex yet.
328+
structured = true;
329+
hasOccurrenceIndex = false;
323330
size = ReadWriteIOUtils.readInt(stream);
324331
} else if (markerOrSize >= 0) {
325332
// Snapshots written before structured duration support started with the CQ count.
326333
structured = false;
334+
hasOccurrenceIndex = false;
327335
size = markerOrSize;
328336
} else {
329337
throw new IOException(
@@ -337,7 +345,7 @@ private void deserialize(InputStream stream) throws IOException {
337345
ManagerMessages.EXCEPTION_NEGATIVE_CQ_SNAPSHOT_ENTRY_COUNT_ARG_38750035, size));
338346
}
339347
for (int i = 0; i < size; i++) {
340-
CQEntry cqEntry = CQEntry.deserialize(stream, structured);
348+
CQEntry cqEntry = CQEntry.deserialize(stream, structured, hasOccurrenceIndex);
341349
cqMap.put(cqEntry.cqId, cqEntry);
342350
}
343351
}
@@ -527,7 +535,8 @@ private void serialize(OutputStream stream) throws IOException {
527535
ReadWriteIOUtils.write(nextOccurrenceIndex, stream);
528536
}
529537

530-
private static CQEntry deserialize(InputStream stream, boolean structured) throws IOException {
538+
private static CQEntry deserialize(
539+
InputStream stream, boolean structured, boolean hasOccurrenceIndex) throws IOException {
531540
String cqId = ReadWriteIOUtils.readString(stream);
532541
long everyInterval = ReadWriteIOUtils.readLong(stream);
533542
long boundaryTime = ReadWriteIOUtils.readLong(stream);
@@ -557,7 +566,7 @@ private static CQEntry deserialize(InputStream stream, boolean structured) throw
557566
boolean boundaryExplicit = structured && ReadWriteIOUtils.readBool(stream);
558567
CQState state = CQState.deserialize(ReadWriteIOUtils.readByte(stream));
559568
long lastExecutionTime = ReadWriteIOUtils.readLong(stream);
560-
long nextOccurrenceIndex = structured ? ReadWriteIOUtils.readLong(stream) : -1;
569+
long nextOccurrenceIndex = hasOccurrenceIndex ? ReadWriteIOUtils.readLong(stream) : -1;
561570
return new CQEntry(
562571
cqId,
563572
everyInterval,

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/CQInfoTest.java‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,8 +37,11 @@
3737

3838
import java.io.File;
3939
import java.io.IOException;
40+
import java.nio.ByteBuffer;
41+
import java.nio.file.Files;
4042
import java.time.ZoneId;
4143
import java.time.ZonedDateTime;
44+
import java.util.Arrays;
4245

4346
import static org.apache.iotdb.db.utils.constant.TestConstant.BASE_OUTPUT_PATH;
4447

@@ -200,6 +203,49 @@ public void testOccurrenceIndexCasFencesOutOfOrderCallbacks() {
200203
.getCode());
201204
}
202205

206+
@Test
207+
public void testPreviousStructuredSnapshotWithoutOccurrenceIndex() throws Exception {
208+
File previousSnapshotDir = new File(BASE_OUTPUT_PATH, "snapshot-previous");
209+
if (previousSnapshotDir.exists()) {
210+
FileUtils.deleteDirectory(previousSnapshotDir);
211+
}
212+
previousSnapshotDir.mkdirs();
213+
214+
CQInfo source = new CQInfo();
215+
TCreateCQReq req =
216+
new TCreateCQReq(
217+
"previousStructuredCq",
218+
1000,
219+
0,
220+
1000,
221+
0,
222+
(byte) 0,
223+
"select 1",
224+
"create cq previousStructuredCq",
225+
"UTC",
226+
"root");
227+
req.setDurationEncodingVersion((short) 1);
228+
req.setEveryDuration(new TCQDuration(0, 1000));
229+
req.setStartOffsetDuration(new TCQDuration(0, 1000));
230+
req.setEndOffsetDuration(new TCQDuration(0, 0));
231+
req.setBoundaryExplicit(true);
232+
source.addCQ(new AddCQPlan(req, "previousStructuredToken", 1000));
233+
source.processTakeSnapshot(previousSnapshotDir);
234+
235+
File snapshotFile = new File(previousSnapshotDir, "cq_info.snapshot");
236+
byte[] snapshot = Files.readAllBytes(snapshotFile.toPath());
237+
ByteBuffer.wrap(snapshot).putInt(-1842801);
238+
// The -1842801 format predates nextOccurrenceIndex, which is the final field of each entry.
239+
Files.write(snapshotFile.toPath(), Arrays.copyOf(snapshot, snapshot.length - Long.BYTES));
240+
241+
CQInfo restored = new CQInfo();
242+
restored.processLoadSnapshot(previousSnapshotDir);
243+
ShowCQResp restoredResponse = restored.showCQ(new ShowCQPlan("previousStructuredCq"));
244+
Assert.assertEquals(1, restoredResponse.getCqList().size());
245+
Assert.assertEquals(-1, restoredResponse.getCqList().get(0).getNextOccurrenceIndex());
246+
FileUtils.deleteDirectory(previousSnapshotDir);
247+
}
248+
203249
@Test
204250
public void testCalendarOccurrenceProgressSurvivesSnapshotRecovery() throws Exception {
205251
ZoneId zone = ZoneId.of("UTC");

0 commit comments

Comments
 (0)