Skip to content

Commit 7026e36

Browse files
DaZuiZuiDaZuiZui
committed
test: cover calendar CQ scheduling recovery cases
1 parent 9c1599b commit 7026e36

3 files changed

Lines changed: 240 additions & 21 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/cq/CQScheduleTask.java‎

Lines changed: 45 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -438,19 +438,16 @@ public AsyncExecuteCQCallback(long startTime, long endTime, long expectedIndex)
438438
}
439439

440440
private long nextOccurrenceIndex(long callbackTime) {
441-
long next = Math.addExact(expectedIndex, 1);
442-
if (timeoutPolicy == TimeoutPolicy.BLOCKED) {
443-
return next;
444-
} else if (timeoutPolicy == TimeoutPolicy.DISCARD) {
445-
long lowerBound =
446-
occurrenceIndex >= 0
447-
? CQCalendarUtils.firstOccurrenceIndex(
448-
boundaryTime, everyDuration, callbackTime, scheduleZone)
449-
: fixedLowerBound(callbackTime);
450-
return Math.max(next, lowerBound);
451-
} else {
452-
throw new IllegalArgumentException(ManagerMessages.UNKNOWN_TIMEOUTPOLICY + timeoutPolicy);
453-
}
441+
return calculateNextOccurrenceIndex(
442+
timeoutPolicy,
443+
expectedIndex,
444+
callbackTime,
445+
executionTime,
446+
everyInterval,
447+
occurrenceIndex,
448+
boundaryTime,
449+
everyDuration,
450+
scheduleZone);
454451
}
455452

456453
private void advanceLegacyExecutionTime(long callbackTime) {
@@ -469,14 +466,6 @@ private void advanceLegacyExecutionTime(long callbackTime) {
469466
}
470467
}
471468

472-
private long fixedLowerBound(long callbackTime) {
473-
if (callbackTime <= executionTime) {
474-
return expectedIndex;
475-
}
476-
return Math.addExact(
477-
expectedIndex, Math.addExact((callbackTime - executionTime - 1) / everyInterval, 1));
478-
}
479-
480469
private void persistProgress(long targetIndex, long callbackTime) {
481470
if (occurrenceIndex < 0) {
482471
TSStatus result;
@@ -572,4 +561,39 @@ public void onError(Exception exception) {
572561
}
573562
}
574563
}
564+
565+
/** Calculates the next occurrence selected after a successful callback. */
566+
static long calculateNextOccurrenceIndex(
567+
TimeoutPolicy timeoutPolicy,
568+
long expectedIndex,
569+
long callbackTime,
570+
long executionTime,
571+
long everyInterval,
572+
long occurrenceIndex,
573+
long boundaryTime,
574+
TimeDuration everyDuration,
575+
ZoneId scheduleZone) {
576+
long next = Math.addExact(expectedIndex, 1);
577+
if (timeoutPolicy == TimeoutPolicy.BLOCKED) {
578+
return next;
579+
}
580+
if (timeoutPolicy == TimeoutPolicy.DISCARD) {
581+
long lowerBound =
582+
occurrenceIndex >= 0
583+
? CQCalendarUtils.firstOccurrenceIndex(
584+
boundaryTime, everyDuration, callbackTime, scheduleZone)
585+
: calculateFixedLowerBound(expectedIndex, callbackTime, executionTime, everyInterval);
586+
return Math.max(next, lowerBound);
587+
}
588+
throw new IllegalArgumentException(ManagerMessages.UNKNOWN_TIMEOUTPOLICY + timeoutPolicy);
589+
}
590+
591+
private static long calculateFixedLowerBound(
592+
long expectedIndex, long callbackTime, long executionTime, long everyInterval) {
593+
if (callbackTime <= executionTime) {
594+
return expectedIndex;
595+
}
596+
return Math.addExact(
597+
expectedIndex, Math.addExact((callbackTime - executionTime - 1) / everyInterval, 1));
598+
}
575599
}

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/cq/CQCalendarUtilsTest.java‎

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.iotdb.confignode.manager.cq;
19+
20+
import org.apache.iotdb.commons.cq.TimeoutPolicy;
21+
import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils;
22+
23+
import org.apache.tsfile.utils.TimeDuration;
24+
import org.junit.Test;
25+
26+
import java.time.Instant;
27+
import java.time.ZoneId;
28+
import java.time.ZonedDateTime;
29+
import java.util.concurrent.TimeUnit;
30+
31+
import static org.junit.Assert.assertEquals;
32+
33+
public class CQCalendarUtilsTest {
34+
35+
private static final ZoneId UTC = ZoneId.of("UTC");
36+
37+
@Test
38+
public void testMonthEndSequenceIsAnchoredToOriginalBoundary() {
39+
long boundary = epochTimestamp(2024, 1, 31, 0, 0, UTC);
40+
TimeDuration every = new TimeDuration(1, 0);
41+
42+
assertEquals(
43+
epochTimestamp(2024, 2, 29, 0, 0, UTC),
44+
CQCalendarUtils.occurrence(boundary, every, 1, UTC));
45+
assertEquals(
46+
epochTimestamp(2024, 3, 31, 0, 0, UTC),
47+
CQCalendarUtils.occurrence(boundary, every, 2, UTC));
48+
assertEquals(
49+
epochTimestamp(2024, 4, 30, 0, 0, UTC),
50+
CQCalendarUtils.occurrence(boundary, every, 3, UTC));
51+
}
52+
53+
@Test
54+
public void testLeapDayYearSequenceReachesNextLeapYear() {
55+
long boundary = epochTimestamp(2020, 2, 29, 0, 0, UTC);
56+
TimeDuration everyYear = new TimeDuration(12, 0);
57+
58+
assertEquals(
59+
epochTimestamp(2021, 2, 28, 0, 0, UTC),
60+
CQCalendarUtils.occurrence(boundary, everyYear, 1, UTC));
61+
assertEquals(
62+
epochTimestamp(2024, 2, 29, 0, 0, UTC),
63+
CQCalendarUtils.occurrence(boundary, everyYear, 4, UTC));
64+
}
65+
66+
@Test
67+
public void testOmittedBoundaryUsesLocalEpochInPersistedZone() {
68+
ZoneId shanghai = ZoneId.of("Asia/Shanghai");
69+
assertEquals(
70+
epochTimestamp(1970, 1, 1, 0, 0, shanghai), CQCalendarUtils.localEpochBoundary(shanghai));
71+
}
72+
73+
@Test
74+
public void testRangeEndpointsAreDerivedFromBoundaryVector() {
75+
long boundary = epochTimestamp(2024, 1, 31, 0, 0, UTC);
76+
TimeDuration every = new TimeDuration(1, 0);
77+
long start = CQCalendarUtils.applyVector(boundary, 1, 0, UTC);
78+
long end = CQCalendarUtils.applyVector(boundary, 2, 0, UTC);
79+
80+
assertEquals(epochTimestamp(2024, 2, 29, 0, 0, UTC), start);
81+
assertEquals(epochTimestamp(2024, 3, 31, 0, 0, UTC), end);
82+
assertEquals(end, CQCalendarUtils.occurrence(boundary, every, 2, UTC));
83+
}
84+
85+
@Test
86+
public void testDstUsesZoneRulesForCalendarAndElapsedParts() {
87+
ZoneId newYork = ZoneId.of("America/New_York");
88+
long monthBoundary = epochTimestamp(2024, 2, 10, 2, 30, newYork);
89+
long monthOccurrence =
90+
CQCalendarUtils.occurrence(monthBoundary, new TimeDuration(1, 0), 1, newYork);
91+
assertEquals(epochTimestamp(2024, 3, 10, 3, 30, newYork), monthOccurrence);
92+
93+
long elapsedBoundary = epochTimestamp(2024, 3, 9, 12, 0, newYork);
94+
long oneDay = TimestampPrecisionUtils.currPrecision.convert(1, TimeUnit.DAYS);
95+
long elapsedOccurrence =
96+
CQCalendarUtils.occurrence(elapsedBoundary, new TimeDuration(0, oneDay), 1, newYork);
97+
assertEquals(epochTimestamp(2024, 3, 10, 13, 0, newYork), elapsedOccurrence);
98+
}
99+
100+
@Test
101+
public void testBlockedKeepsTheNextOccurrenceWhileDiscardSkipsMissedOccurrences() {
102+
long boundary = epochTimestamp(2024, 1, 1, 0, 0, UTC);
103+
TimeDuration every = new TimeDuration(1, 0);
104+
long callbackTime = epochTimestamp(2024, 1, 4, 12, 0, UTC);
105+
long executionTime = CQCalendarUtils.occurrence(boundary, every, 1, UTC);
106+
107+
assertEquals(
108+
2,
109+
CQScheduleTask.calculateNextOccurrenceIndex(
110+
TimeoutPolicy.BLOCKED, 1, callbackTime, executionTime, 0, 1, boundary, every, UTC));
111+
assertEquals(
112+
4,
113+
CQScheduleTask.calculateNextOccurrenceIndex(
114+
TimeoutPolicy.DISCARD, 1, callbackTime, executionTime, 0, 1, boundary, every, UTC));
115+
}
116+
117+
@Test
118+
public void testDiscardNeverMovesBeforeTheCurrentOccurrence() {
119+
long boundary = epochTimestamp(2024, 1, 1, 0, 0, UTC);
120+
TimeDuration every = new TimeDuration(1, 0);
121+
long currentOccurrence = CQCalendarUtils.occurrence(boundary, every, 2, UTC);
122+
123+
assertEquals(
124+
3,
125+
CQScheduleTask.calculateNextOccurrenceIndex(
126+
TimeoutPolicy.DISCARD,
127+
2,
128+
currentOccurrence,
129+
currentOccurrence,
130+
0,
131+
2,
132+
boundary,
133+
every,
134+
UTC));
135+
}
136+
137+
private static long epochTimestamp(
138+
int year, int month, int day, int hour, int minute, ZoneId zone) {
139+
Instant instant = ZonedDateTime.of(year, month, day, hour, minute, 0, 0, zone).toInstant();
140+
return Math.addExact(
141+
TimestampPrecisionUtils.currPrecision.convert(instant.getEpochSecond(), TimeUnit.SECONDS),
142+
TimestampPrecisionUtils.currPrecision.convert(instant.getNano(), TimeUnit.NANOSECONDS));
143+
}
144+
145+
private static long epochTimestamp(int year, int month, int day, int hour, ZoneId zone) {
146+
return epochTimestamp(year, month, day, hour, 0, zone);
147+
}
148+
}

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

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

3838
import java.io.File;
3939
import java.io.IOException;
40+
import java.time.ZoneId;
41+
import java.time.ZonedDateTime;
4042

4143
import static org.apache.iotdb.db.utils.constant.TestConstant.BASE_OUTPUT_PATH;
4244

@@ -197,4 +199,49 @@ public void testOccurrenceIndexCasFencesOutOfOrderCallbacks() {
197199
new UpdateCQLastExecTimePlan("indexedCq", 2000, "indexedToken", 1, 2))
198200
.getCode());
199201
}
202+
203+
@Test
204+
public void testCalendarOccurrenceProgressSurvivesSnapshotRecovery() throws Exception {
205+
ZoneId zone = ZoneId.of("UTC");
206+
long boundary = ZonedDateTime.of(2024, 1, 31, 0, 0, 0, 0, zone).toInstant().toEpochMilli();
207+
long firstOccurrence =
208+
ZonedDateTime.of(2024, 2, 29, 0, 0, 0, 0, zone).toInstant().toEpochMilli();
209+
TCreateCQReq req =
210+
new TCreateCQReq(
211+
"snapshotCalendarCq",
212+
0,
213+
boundary,
214+
0,
215+
0,
216+
(byte) 0,
217+
"select 1",
218+
"create cq snapshotCalendarCq",
219+
"UTC",
220+
"root");
221+
req.setDurationEncodingVersion((short) 1);
222+
req.setEveryDuration(new TCQDuration(1, 0));
223+
req.setStartOffsetDuration(new TCQDuration(1, 0));
224+
req.setEndOffsetDuration(new TCQDuration(0, 0));
225+
req.setBoundaryExplicit(true);
226+
227+
Assert.assertEquals(
228+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
229+
cqInfo.addCQ(new AddCQPlan(req, "snapshotCalendarToken", firstOccurrence)).getCode());
230+
Assert.assertEquals(
231+
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
232+
cqInfo
233+
.updateCQLastExecutionTime(
234+
new UpdateCQLastExecTimePlan(
235+
"snapshotCalendarCq", firstOccurrence, "snapshotCalendarToken", 1, 2))
236+
.getCode());
237+
238+
cqInfo.processTakeSnapshot(snapshotDir);
239+
CQInfo restored = new CQInfo();
240+
restored.processLoadSnapshot(snapshotDir);
241+
242+
ShowCQResp response = restored.showCQ(new ShowCQPlan("snapshotCalendarCq"));
243+
Assert.assertEquals(1, response.getCqList().size());
244+
Assert.assertEquals(2, response.getCqList().get(0).getNextOccurrenceIndex());
245+
Assert.assertEquals(firstOccurrence, response.getCqList().get(0).getLastExecutionTime());
246+
}
200247
}

0 commit comments

Comments
 (0)