Skip to content

Commit 9c1599b

Browse files
DaZuiZuiDaZuiZui
committed
Fix calendar CQ scheduling compatibility and progress
1 parent e7a77e1 commit 9c1599b

17 files changed

Lines changed: 584 additions & 93 deletions

File tree

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -714,5 +714,21 @@ private ManagerMessages() {}
714714
public static final String
715715
MESSAGE_CQ_LEGACY_DURATION_FIELDS_CONFLICT_WITH_STRUCTURED_DURATION_FIELDS_4D6C6D67 =
716716
"CQ legacy duration fields conflict with structured duration fields";
717+
public static final String MESSAGE_CQ_CALENDAR_DURATION_REQUIRES_ALL_NODES_SUPPORT_49534072 =
718+
"CQ calendar duration requires all cluster nodes to support duration encoding version 1";
719+
public static final String MESSAGE_CQ_DURATION_ENCODING_MARKER_REQUIRED_9035980A =
720+
"CQ duration encoding marker is required for new requests";
721+
public static final String MESSAGE_CQ_DOES_NOT_HAVE_OCCURRENCE_INDEX_METADATA_929A7F0C =
722+
"CQ does not have occurrence-index metadata";
723+
public static final String MESSAGE_CQ_OCCURRENCE_CALLBACK_IS_STALE_36C5FBFC =
724+
"CQ occurrence callback is stale";
725+
public static final String MESSAGE_CQ_OCCURRENCE_INDEX_IS_AHEAD_OF_THE_CALLBACK_8A18ECC9 =
726+
"CQ occurrence index is ahead of the callback";
727+
public static final String EXCEPTION_INVALID_CQ_OCCURRENCE_INDEX_TRANSITION_AC6BFC4D =
728+
"Invalid CQ occurrence index transition";
729+
public static final String EXCEPTION_UNSUPPORTED_CQ_SNAPSHOT_VERSION_ARG_0F1E0A01 =
730+
"Unsupported CQ snapshot version: %d";
731+
public static final String EXCEPTION_NEGATIVE_CQ_SNAPSHOT_ENTRY_COUNT_ARG_38750035 =
732+
"Negative CQ snapshot entry count: %d";
717733

718734
}

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -693,5 +693,21 @@ private ManagerMessages() {}
693693
public static final String
694694
MESSAGE_CQ_LEGACY_DURATION_FIELDS_CONFLICT_WITH_STRUCTURED_DURATION_FIELDS_4D6C6D67 =
695695
"CQ legacy duration fields 与结构化 duration 字段冲突";
696+
public static final String MESSAGE_CQ_CALENDAR_DURATION_REQUIRES_ALL_NODES_SUPPORT_49534072 =
697+
"CQ 日历 duration 要求集群所有节点支持 duration encoding version 1";
698+
public static final String MESSAGE_CQ_DURATION_ENCODING_MARKER_REQUIRED_9035980A =
699+
"新建 CQ 请求必须包含 duration encoding marker";
700+
public static final String MESSAGE_CQ_DOES_NOT_HAVE_OCCURRENCE_INDEX_METADATA_929A7F0C =
701+
"CQ 没有 occurrence-index 元数据";
702+
public static final String MESSAGE_CQ_OCCURRENCE_CALLBACK_IS_STALE_36C5FBFC =
703+
"CQ occurrence callback 已过期";
704+
public static final String MESSAGE_CQ_OCCURRENCE_INDEX_IS_AHEAD_OF_THE_CALLBACK_8A18ECC9 =
705+
"CQ occurrence index 超前于 callback";
706+
public static final String EXCEPTION_INVALID_CQ_OCCURRENCE_INDEX_TRANSITION_AC6BFC4D =
707+
"无效的 CQ occurrence index 转换";
708+
public static final String EXCEPTION_UNSUPPORTED_CQ_SNAPSHOT_VERSION_ARG_0F1E0A01 =
709+
"不支持的 CQ snapshot 版本：%d";
710+
public static final String EXCEPTION_NEGATIVE_CQ_SNAPSHOT_ENTRY_COUNT_ARG_38750035 =
711+
"CQ snapshot 条目数量不能为负数：%d";
696712

697713
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/confignode/UpdateVersionInfoPlan.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,14 @@ protected void serializeImpl(DataOutputStream stream) throws IOException {
5959
ReadWriteIOUtils.write(nodeId, stream);
6060
ReadWriteIOUtils.write(versionInfo.getVersion(), stream);
6161
ReadWriteIOUtils.write(versionInfo.getBuildInfo(), stream);
62+
// Optional tail keeps this plan readable by pre-calendar-duration ConfigNodes.
63+
ReadWriteIOUtils.write(versionInfo.isSetSupportedCQDurationEncodingVersions(), stream);
64+
if (versionInfo.isSetSupportedCQDurationEncodingVersions()) {
65+
ReadWriteIOUtils.write(versionInfo.getSupportedCQDurationEncodingVersions().size(), stream);
66+
for (short version : versionInfo.getSupportedCQDurationEncodingVersions()) {
67+
ReadWriteIOUtils.write(version, stream);
68+
}
69+
}
6270
}
6371

6472
@Override
@@ -67,6 +75,17 @@ protected void deserializeImpl(ByteBuffer buffer) {
6775
versionInfo =
6876
new TNodeVersionInfo(
6977
ReadWriteIOUtils.readString(buffer), ReadWriteIOUtils.readString(buffer));
78+
if (buffer.hasRemaining()) {
79+
boolean hasCapabilities = ReadWriteIOUtils.readBool(buffer);
80+
if (hasCapabilities) {
81+
int size = ReadWriteIOUtils.readInt(buffer);
82+
java.util.Set<Short> capabilities = new java.util.HashSet<>();
83+
for (int i = 0; i < size; i++) {
84+
capabilities.add(ReadWriteIOUtils.readShort(buffer));
85+
}
86+
versionInfo.setSupportedCQDurationEncodingVersions(capabilities);
87+
}
88+
}
7089
}
7190

7291
@Override

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/cq/UpdateCQLastExecTimePlan.java‎

Lines changed: 52 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.apache.iotdb.confignode.consensus.request.write.cq;
2121

2222
import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
23+
import org.apache.iotdb.confignode.i18n.ManagerMessages;
2324

2425
import org.apache.tsfile.external.commons.lang3.Validate;
2526
import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -39,6 +40,10 @@ public class UpdateCQLastExecTimePlan extends ConfigPhysicalPlan {
3940

4041
private String cqToken;
4142

43+
private boolean hasOccurrenceIndex;
44+
private long expectedIndex;
45+
private long targetIndex;
46+
4247
public UpdateCQLastExecTimePlan() {
4348
super(UPDATE_CQ_LAST_EXEC_TIME);
4449
}
@@ -52,6 +57,18 @@ public UpdateCQLastExecTimePlan(String cqId, long executionTime, String cqToken)
5257
this.cqToken = cqToken;
5358
}
5459

60+
public UpdateCQLastExecTimePlan(
61+
String cqId, long executionTime, String cqToken, long expectedIndex, long targetIndex) {
62+
this(cqId, executionTime, cqToken);
63+
if (expectedIndex < 0 || targetIndex <= expectedIndex) {
64+
throw new IllegalArgumentException(
65+
ManagerMessages.EXCEPTION_INVALID_CQ_OCCURRENCE_INDEX_TRANSITION_AC6BFC4D);
66+
}
67+
this.hasOccurrenceIndex = true;
68+
this.expectedIndex = expectedIndex;
69+
this.targetIndex = targetIndex;
70+
}
71+
5572
public String getCqId() {
5673
return cqId;
5774
}
@@ -64,19 +81,43 @@ public String getCqToken() {
6481
return cqToken;
6582
}
6683

84+
public boolean hasOccurrenceIndex() {
85+
return hasOccurrenceIndex;
86+
}
87+
88+
public long getExpectedIndex() {
89+
return expectedIndex;
90+
}
91+
92+
public long getTargetIndex() {
93+
return targetIndex;
94+
}
95+
6796
@Override
6897
protected void serializeImpl(DataOutputStream stream) throws IOException {
6998
stream.writeShort(getType().getPlanType());
7099
ReadWriteIOUtils.write(cqId, stream);
71100
ReadWriteIOUtils.write(executionTime, stream);
72101
ReadWriteIOUtils.write(cqToken, stream);
102+
ReadWriteIOUtils.write(hasOccurrenceIndex, stream);
103+
if (hasOccurrenceIndex) {
104+
ReadWriteIOUtils.write(expectedIndex, stream);
105+
ReadWriteIOUtils.write(targetIndex, stream);
106+
}
73107
}
74108

75109
@Override
76110
protected void deserializeImpl(ByteBuffer buffer) throws IOException {
77111
cqId = ReadWriteIOUtils.readString(buffer);
78112
executionTime = ReadWriteIOUtils.readLong(buffer);
79113
cqToken = ReadWriteIOUtils.readString(buffer);
114+
if (buffer.hasRemaining()) {
115+
hasOccurrenceIndex = ReadWriteIOUtils.readBool(buffer);
116+
if (hasOccurrenceIndex) {
117+
expectedIndex = ReadWriteIOUtils.readLong(buffer);
118+
targetIndex = ReadWriteIOUtils.readLong(buffer);
119+
}
120+
}
80121
}
81122

82123
@Override
@@ -92,12 +133,22 @@ public boolean equals(Object o) {
92133
}
93134
UpdateCQLastExecTimePlan that = (UpdateCQLastExecTimePlan) o;
94135
return executionTime == that.executionTime
136+
&& hasOccurrenceIndex == that.hasOccurrenceIndex
137+
&& expectedIndex == that.expectedIndex
138+
&& targetIndex == that.targetIndex
95139
&& cqId.equals(that.cqId)
96140
&& cqToken.equals(that.cqToken);
97141
}
98142

99143
@Override
100144
public int hashCode() {
101-
return Objects.hash(super.hashCode(), cqId, executionTime, cqToken);
145+
return Objects.hash(
146+
super.hashCode(),
147+
cqId,
148+
executionTime,
149+
cqToken,
150+
hasOccurrenceIndex,
151+
expectedIndex,
152+
targetIndex);
102153
}
103154
}

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

Lines changed: 64 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.apache.iotdb.confignode.persistence.cq.CQInfo;
3434
import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq;
3535
import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq;
36+
import org.apache.iotdb.confignode.rpc.thrift.TNodeVersionInfo;
3637
import org.apache.iotdb.confignode.rpc.thrift.TShowCQResp;
3738
import org.apache.iotdb.consensus.common.DataSet;
3839
import org.apache.iotdb.consensus.exception.ConsensusException;
@@ -90,7 +91,16 @@ public TSStatus createCQ(TCreateCQReq req) {
9091

9192
private TSStatus validateDurationEncoding(TCreateCQReq req) {
9293
if (!req.isSetDurationEncodingVersion()) {
93-
// Requests from older DataNodes contain only the legacy fixed-duration fields.
94+
// Preserve wire compatibility for legacy fixed-duration clients. A request that starts
95+
// sending structured fields must include the version marker so it cannot be misread by an
96+
// older ConfigNode.
97+
if (req.isSetEveryDuration()
98+
|| req.isSetStartOffsetDuration()
99+
|| req.isSetEndOffsetDuration()
100+
|| req.isSetBoundaryExplicit()) {
101+
return new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode())
102+
.setMessage(ManagerMessages.MESSAGE_CQ_DURATION_ENCODING_MARKER_REQUIRED_9035980A);
103+
}
94104
return null;
95105
}
96106
if (req.getDurationEncodingVersion() != 1
@@ -112,6 +122,16 @@ private TSStatus validateDurationEncoding(TCreateCQReq req) {
112122
return new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode())
113123
.setMessage(ManagerMessages.MESSAGE_CQ_DURATIONS_MUST_BE_NON_NEGATIVE_BE23CE04);
114124
}
125+
if (req.getEveryDuration().getMonthPart() > Integer.MAX_VALUE
126+
|| req.getStartOffsetDuration().getMonthPart() > Integer.MAX_VALUE
127+
|| req.getEndOffsetDuration().getMonthPart() > Integer.MAX_VALUE) {
128+
return new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode())
129+
.setMessage(ManagerMessages.MESSAGE_CQ_DURATIONS_MUST_BE_NON_NEGATIVE_BE23CE04);
130+
}
131+
boolean hasCalendarDuration =
132+
req.getEveryDuration().getMonthPart() != 0
133+
|| req.getStartOffsetDuration().getMonthPart() != 0
134+
|| req.getEndOffsetDuration().getMonthPart() != 0;
115135
if ((req.getEveryDuration().getMonthPart() == 0
116136
&& req.everyInterval != req.getEveryDuration().getNonMonthDuration())
117137
|| (req.getStartOffsetDuration().getMonthPart() == 0
@@ -120,15 +140,57 @@ private TSStatus validateDurationEncoding(TCreateCQReq req) {
120140
&& req.endTimeOffset != req.getEndOffsetDuration().getNonMonthDuration())
121141
|| (req.getEveryDuration().getMonthPart() != 0 && req.everyInterval != 0)
122142
|| (req.getStartOffsetDuration().getMonthPart() != 0 && req.startTimeOffset != 0)
123-
|| (req.getEndOffsetDuration().getMonthPart() != 0 && req.endTimeOffset != 0)) {
143+
|| (req.getEndOffsetDuration().getMonthPart() != 0 && req.endTimeOffset != 0)
144+
|| (hasCalendarDuration
145+
&& (req.everyInterval != 0 || req.startTimeOffset != 0 || req.endTimeOffset != 0))) {
124146
return new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode())
125147
.setMessage(
126148
ManagerMessages
127149
.MESSAGE_CQ_LEGACY_DURATION_FIELDS_CONFLICT_WITH_STRUCTURED_DURATION_FIELDS_4D6C6D67);
128150
}
151+
if (hasCalendarDuration && !allClusterNodesSupportDurationEncodingV1()) {
152+
return new TSStatus(TSStatusCode.SEMANTIC_ERROR.getStatusCode())
153+
.setMessage(
154+
ManagerMessages.MESSAGE_CQ_CALENDAR_DURATION_REQUIRES_ALL_NODES_SUPPORT_49534072);
155+
}
129156
return null;
130157
}
131158

159+
private boolean allClusterNodesSupportDurationEncodingV1() {
160+
java.util.Map<Integer, TNodeVersionInfo> versionInfo =
161+
configManager.getNodeManager().getNodeVersionInfo();
162+
if (versionInfo.isEmpty()) {
163+
return false;
164+
}
165+
boolean hasRegisteredNode = false;
166+
// Check every registered node explicitly. A missing heartbeat/version entry must not allow a
167+
// calendar CQ to be created during a rolling upgrade.
168+
for (org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation node :
169+
configManager.getNodeManager().getRegisteredConfigNodes()) {
170+
hasRegisteredNode = true;
171+
if (!supportsDurationEncodingV1(versionInfo.get(node.getConfigNodeId()))) {
172+
return false;
173+
}
174+
}
175+
for (org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration node :
176+
configManager.getNodeManager().getRegisteredDataNodes()) {
177+
hasRegisteredNode = true;
178+
if (!supportsDurationEncodingV1(versionInfo.get(node.getLocation().getDataNodeId()))) {
179+
return false;
180+
}
181+
}
182+
return hasRegisteredNode;
183+
}
184+
185+
private boolean supportsDurationEncodingV1(TNodeVersionInfo info) {
186+
if (info == null
187+
|| !info.isSetSupportedCQDurationEncodingVersions()
188+
|| !info.getSupportedCQDurationEncodingVersions().contains((short) 1)) {
189+
return false;
190+
}
191+
return true;
192+
}
193+
132194
public TSStatus dropCQ(TDropCQReq req) {
133195
lock.readLock().lock();
134196
try {

0 commit comments

Comments
 (0)