feat/cq-calendar-month-intervals

integration-test/src/test/java/org/apache/iotdb/confignode/it/IoTDBConfigNodeSnapshotIT.java

@@ -32,6 +32,7 @@
 import org.apache.iotdb.commons.trigger.TriggerInformation;
 import org.apache.iotdb.commons.trigger.service.TriggerExecutableManager;
 import org.apache.iotdb.commons.udf.UDFInformation;
+import org.apache.iotdb.confignode.rpc.thrift.TCQDuration;
 import org.apache.iotdb.confignode.rpc.thrift.TCQEntry;
 import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq;
 import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq;
@@ -354,6 +355,14 @@ private Set<TCQEntry> createCQs(SyncConfigNodeIServiceClient client) throws TExc
             "UTC",
             "root");
 
+    for (TCreateCQReq req : new TCreateCQReq[] {req1, req2}) {
+      req.setDurationEncodingVersion((short) 1);
+      req.setEveryDuration(new TCQDuration(0, 1000));
+      req.setStartOffsetDuration(new TCQDuration(0, 1000));
+      req.setEndOffsetDuration(new TCQDuration(0, 0));
+      req.setBoundaryExplicit(true);
+    }
+
     assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.createCQ(req1).getCode());
     assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), client.createCQ(req2).getCode());
 

integration-test/src/test/java/org/apache/iotdb/db/it/cq/IoTDBCQExecIT.java

@@ -32,6 +32,9 @@
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.Statement;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.util.concurrent.TimeUnit;
 
 import static org.apache.iotdb.itbase.constant.TestConstant.TIMESTAMP_STR;
@@ -52,6 +55,89 @@ public static void tearDown() throws Exception {
     EnvFactory.getEnv().cleanClusterEnvironment();
   }
 
+  @Test
+  public void testCalendarMonthCQExecutionUsesNaturalMonthWindow() {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      connection.setClientInfo("time_zone", "UTC");
+      long now = System.currentTimeMillis();
+      long firstExecutionTime = now + 30_000;
+      ZoneId utc = ZoneId.of("UTC");
+      long calendarStart =
+          ZonedDateTime.ofInstant(Instant.ofEpochMilli(firstExecutionTime), utc)
+              .minusMonths(1)
+              .toInstant()
+              .toEpochMilli();
+      long thirtyDayStart = firstExecutionTime - TimeUnit.DAYS.toMillis(30);
+
+      statement.execute("create timeseries root.sg.calendar.s1 WITH DATATYPE=INT64");
+      statement.execute("create timeseries root.sg.calendar.s1_max WITH DATATYPE=INT64");
+      statement.execute("INSERT INTO root.sg.calendar(time, s1) VALUES (0,0)");
+      statement.execute(
+          String.format(
+              "INSERT INTO root.sg.calendar(time, s1) VALUES (%d, 777), (%d, 100), (%d, 10), (%d, 999)",
+              calendarStart - 1, calendarStart, firstExecutionTime - 1, firstExecutionTime));
+      // On 28/29-day months the 30-day window starts before the natural month. A 30d RANGE
+      // would include this 888; a calendar RANGE 1mo must not.
+      if (thirtyDayStart < calendarStart - 1) {
+        statement.execute(
+            String.format(
+                "INSERT INTO root.sg.calendar(time, s1) VALUES (%d, 888)", thirtyDayStart));
+      }
+
+      statement.execute(
+          "CREATE CONTINUOUS QUERY cq_calendar_month\n"
+              + "RESAMPLE EVERY 1mo\n"
+              + String.format("BOUNDARY %d\n", firstExecutionTime)
+              + "RANGE 1mo\n"
+              + "BEGIN\n"
+              + "  SELECT max_value(s1)\n"
+              + "  INTO root.sg.calendar(s1_max)\n"
+              + "  FROM root.sg.calendar\n"
+              + "  GROUP BY(1mo)\n"
+              + "END");
+
+      if (System.currentTimeMillis() > firstExecutionTime) {
+        // Do not return silently: a vacuous pass would hide a regression in the calendar window.
+        statement.execute("DROP CQ cq_calendar_month");
+        fail("test setup exceeded the scheduled first execution time; increase the margin");
+      }
+
+      long targetTime = firstExecutionTime + 10_000;
+      while (System.currentTimeMillis() - targetTime < 0) {
+        TimeUnit.SECONDS.sleep(1);
+      }
+
+      try (ResultSet resultSet = statement.executeQuery("select s1_max from root.sg.calendar")) {
+        boolean sawNaturalMonthBucket = false;
+        while (resultSet.next()) {
+          long time = resultSet.getLong(TIMESTAMP_STR);
+          long value = resultSet.getLong("root.sg.calendar.s1_max");
+          assertEquals(
+              "CQ RANGE must stay inside the just-finished natural month",
+              true,
+              time >= calendarStart && time < firstExecutionTime);
+          assertEquals(
+              "points before/after the natural month must not contribute", true, value != 777);
+          assertEquals("the BOUNDARY instant is exclusive", true, value != 999);
+          assertEquals("a 30-day-only point must not become the monthly max", true, value != 888);
+          if (time == calendarStart && value == 100) {
+            sawNaturalMonthBucket = true;
+          }
+        }
+        assertEquals(
+            "the first GROUP BY month bucket should start at BOUNDARY-1mo",
+            true,
+            sawNaturalMonthBucket);
+      } finally {
+        statement.execute("DROP CQ cq_calendar_month");
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+      fail(e.getMessage());
+    }
+  }
+
   @Test
   public void testCQExecution1() {
     String insertTemplate =

integration-test/src/test/java/org/apache/iotdb/db/it/cq/IoTDBCQIT.java

@@ -36,6 +36,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 @RunWith(IoTDBTestRunner.class)
@@ -407,6 +408,63 @@ public void testCreateCorrectCQ() {
     }
   }
 
+  @Test
+  public void testCreateCalendarMonthCQ() {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      statement.execute(
+          "CREATE CQ calendar_cq_1\n"
+              + "RESAMPLE EVERY 1mo RANGE 1mo\n"
+              + "BEGIN\n"
+              + "  SELECT max_value(s1)\n"
+              + "    INTO root.sg.d1(s1_max)\n"
+              + "    FROM root.sg.d1\n"
+              + "    GROUP BY(1mo)\n"
+              + "END");
+      try (ResultSet resultSet = statement.executeQuery("show CQS")) {
+        boolean found = false;
+        while (resultSet.next()) {
+          if ("calendar_cq_1".equals(resultSet.getString(1))) {
+            found = true;
+            assertEquals("ACTIVE", resultSet.getString(3));
+          }
+        }
+        assertTrue(found);
+      }
+      statement.execute("DROP CQ calendar_cq_1");
+    } catch (Exception e) {
+      e.printStackTrace();
+      fail(e.getMessage());
+    }
+  }
+
+  @Test
+  public void testRejectIncomparableCalendarAndFixedCQDurations() {
+    try (Connection connection = EnvFactory.getEnv().getConnection();
+        Statement statement = connection.createStatement()) {
+      try {
+        statement.execute(
+            "CREATE CQ calendar_cq_incomparable\n"
+                + "RESAMPLE EVERY 1mo RANGE 30d\n"
+                + "BEGIN\n"
+                + "  SELECT max_value(s1)\n"
+                + "    INTO root.sg.d1(s1_max)\n"
+                + "    FROM root.sg.d1\n"
+                + "    GROUP BY(1mo)\n"
+                + "END");
+        fail();
+      } catch (Exception e) {
+        assertEquals(
+            TSStatusCode.SEMANTIC_ERROR.getStatusCode()
+                + ": CQ: The start time offset should be greater than or equal to every interval.",
+            e.getMessage());
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+      fail(e.getMessage());
+    }
+  }
+
   // =======================================show cq======================================
   @Test
   public void testShowCQ() {

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

@@ -702,4 +702,44 @@ private ManagerMessages() {}
       MESSAGE_ARG_PLEASE_MANUALLY_CHECK_LATER_WHETHER_THE_PROCEDURE_IS_EXECUTED_SUCCESSFULLY_A82B739D =
           "%s Please manually check later whether the procedure is executed successfully.";
 
+  public static final String EXCEPTION_CQ_EVERY_DURATION_MUST_BE_POSITIVE_69C29D26 =
+      "CQ EVERY duration must be positive";
+  public static final String MESSAGE_CQ_START_OFFSET_MUST_BE_POSITIVE_B837C4F5 =
+      "CQ start offset must be positive";
+  public static final String MESSAGE_CQ_END_OFFSET_MUST_BE_NON_NEGATIVE_64171164 =
+      "CQ end offset must be non-negative";
+  public static final String MESSAGE_CQ_START_OFFSET_MUST_BE_GREATER_THAN_END_OFFSET_5924C189 =
+      "CQ start offset must be greater than end offset";
+  public static final String
+      MESSAGE_CQ_START_OFFSET_MUST_BE_GREATER_THAN_OR_EQUAL_TO_EVERY_DURATION_89628D43 =
+          "CQ start offset must be greater than or equal to EVERY duration";
+  public static final String EXCEPTION_CQ_TIMESTAMP_OVERFLOWS_CONFIGURED_PRECISION_F5FB230C =
+      "CQ timestamp overflows configured precision";
+  public static final String
+      MESSAGE_INVALID_CQ_DURATION_ENCODING_VERSION_1_REQUIRES_ALL_STRUCTURED_FIELDS_FEAD7F92 =
+          "Invalid CQ duration encoding; version 1 requires all structured fields";
+  public static final String MESSAGE_CQ_DURATIONS_MUST_BE_NON_NEGATIVE_BE23CE04 =
+      "CQ durations must be non-negative";
+  public static final String
+      MESSAGE_CQ_LEGACY_DURATION_FIELDS_CONFLICT_WITH_STRUCTURED_DURATION_FIELDS_4D6C6D67 =
+          "CQ legacy duration fields conflict with structured duration fields";
+  public static final String MESSAGE_CQ_CALENDAR_DURATION_REQUIRES_ALL_NODES_SUPPORT_49534072 =
+      "CQ calendar duration requires all cluster nodes to support duration encoding version 1";
+  public static final String MESSAGE_CQ_DURATION_ENCODING_MARKER_REQUIRED_9035980A =
+      "CQ duration encoding marker is required for new requests";
+  public static final String MESSAGE_CQ_DOES_NOT_HAVE_OCCURRENCE_INDEX_METADATA_929A7F0C =
+      "CQ does not have occurrence-index metadata";
+  public static final String MESSAGE_CQ_OCCURRENCE_CALLBACK_IS_STALE_36C5FBFC =
+      "CQ occurrence callback is stale";
+  public static final String MESSAGE_CQ_OCCURRENCE_INDEX_IS_AHEAD_OF_THE_CALLBACK_8A18ECC9 =
+      "CQ occurrence index is ahead of the callback";
+  public static final String EXCEPTION_INVALID_CQ_OCCURRENCE_INDEX_TRANSITION_AC6BFC4D =
+      "Invalid CQ occurrence index transition";
+  public static final String EXCEPTION_NEGATIVE_CQ_SNAPSHOT_ENTRY_COUNT_ARG_38750035 =
+      "Negative CQ snapshot entry count: %d";
+  public static final String EXCEPTION_CQ_OCCURRENCE_INDEX_DOES_NOT_MATCH_EXECUTION_TIME_B2DE4B0F =
+      "CQ occurrence index does not match the scheduled execution time";
+  public static final String EXCEPTION_CQ_RANGE_END_MUST_BE_GREATER_THAN_START_3C91E8B4 =
+      "CQ RANGE end time must be greater than start time";
+
 }

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

@@ -681,4 +681,44 @@ private ManagerMessages() {}
       MESSAGE_ARG_PLEASE_MANUALLY_CHECK_LATER_WHETHER_THE_PROCEDURE_IS_EXECUTED_SUCCESSFULLY_A82B739D =
           "%s 请稍后手动检查该 Procedure 是否执行成功。";
 
+  public static final String EXCEPTION_CQ_EVERY_DURATION_MUST_BE_POSITIVE_69C29D26 =
+      "CQ EVERY duration 必须为正数";
+  public static final String MESSAGE_CQ_START_OFFSET_MUST_BE_POSITIVE_B837C4F5 =
+      "CQ start offset 必须为正数";
+  public static final String MESSAGE_CQ_END_OFFSET_MUST_BE_NON_NEGATIVE_64171164 =
+      "CQ end offset 必须为非负数";
+  public static final String MESSAGE_CQ_START_OFFSET_MUST_BE_GREATER_THAN_END_OFFSET_5924C189 =
+      "CQ start offset 必须大于 end offset";
+  public static final String
+      MESSAGE_CQ_START_OFFSET_MUST_BE_GREATER_THAN_OR_EQUAL_TO_EVERY_DURATION_89628D43 =
+          "CQ start offset 必须大于或等于 EVERY duration";
+  public static final String EXCEPTION_CQ_TIMESTAMP_OVERFLOWS_CONFIGURED_PRECISION_F5FB230C =
+      "CQ timestamp 超出配置的精度范围";
+  public static final String
+      MESSAGE_INVALID_CQ_DURATION_ENCODING_VERSION_1_REQUIRES_ALL_STRUCTURED_FIELDS_FEAD7F92 =
+          "无效的 CQ duration encoding；版本 1 需要所有结构化字段";
+  public static final String MESSAGE_CQ_DURATIONS_MUST_BE_NON_NEGATIVE_BE23CE04 =
+      "CQ duration 必须为非负数";
+  public static final String
+      MESSAGE_CQ_LEGACY_DURATION_FIELDS_CONFLICT_WITH_STRUCTURED_DURATION_FIELDS_4D6C6D67 =
+          "CQ legacy duration fields 与结构化 duration 字段冲突";
+  public static final String MESSAGE_CQ_CALENDAR_DURATION_REQUIRES_ALL_NODES_SUPPORT_49534072 =
+      "CQ 日历 duration 要求集群所有节点支持 duration encoding version 1";
+  public static final String MESSAGE_CQ_DURATION_ENCODING_MARKER_REQUIRED_9035980A =
+      "新建 CQ 请求必须包含 duration encoding marker";
+  public static final String MESSAGE_CQ_DOES_NOT_HAVE_OCCURRENCE_INDEX_METADATA_929A7F0C =
+      "CQ 没有 occurrence-index 元数据";
+  public static final String MESSAGE_CQ_OCCURRENCE_CALLBACK_IS_STALE_36C5FBFC =
+      "CQ occurrence callback 已过期";
+  public static final String MESSAGE_CQ_OCCURRENCE_INDEX_IS_AHEAD_OF_THE_CALLBACK_8A18ECC9 =
+      "CQ occurrence index 超前于 callback";
+  public static final String EXCEPTION_INVALID_CQ_OCCURRENCE_INDEX_TRANSITION_AC6BFC4D =
+      "无效的 CQ occurrence index 转换";
+  public static final String EXCEPTION_NEGATIVE_CQ_SNAPSHOT_ENTRY_COUNT_ARG_38750035 =
+      "CQ snapshot 条目数量不能为负数：%d";
+  public static final String EXCEPTION_CQ_OCCURRENCE_INDEX_DOES_NOT_MATCH_EXECUTION_TIME_B2DE4B0F =
+      "CQ occurrence index 与计划执行时间不匹配";
+  public static final String EXCEPTION_CQ_RANGE_END_MUST_BE_GREATER_THAN_START_3C91E8B4 =
+      "CQ RANGE 结束时间必须大于开始时间";
+
 }

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

@@ -59,6 +59,14 @@ protected void serializeImpl(DataOutputStream stream) throws IOException {
     ReadWriteIOUtils.write(nodeId, stream);
     ReadWriteIOUtils.write(versionInfo.getVersion(), stream);
     ReadWriteIOUtils.write(versionInfo.getBuildInfo(), stream);
+    // Optional tail keeps this plan readable by pre-calendar-duration ConfigNodes.
+    ReadWriteIOUtils.write(versionInfo.isSetSupportedCQDurationEncodingVersions(), stream);
+    if (versionInfo.isSetSupportedCQDurationEncodingVersions()) {
+      ReadWriteIOUtils.write(versionInfo.getSupportedCQDurationEncodingVersions().size(), stream);
+      for (short version : versionInfo.getSupportedCQDurationEncodingVersions()) {
+        ReadWriteIOUtils.write(version, stream);
+      }
+    }
   }
 
   @Override
@@ -67,6 +75,17 @@ protected void deserializeImpl(ByteBuffer buffer) {
     versionInfo =
         new TNodeVersionInfo(
             ReadWriteIOUtils.readString(buffer), ReadWriteIOUtils.readString(buffer));
+    if (buffer.hasRemaining()) {
+      boolean hasCapabilities = ReadWriteIOUtils.readBool(buffer);
+      if (hasCapabilities) {
+        int size = ReadWriteIOUtils.readInt(buffer);
+        java.util.Set<Short> capabilities = new java.util.HashSet<>();
+        for (int i = 0; i < size; i++) {
+          capabilities.add(ReadWriteIOUtils.readShort(buffer));
+        }
+        versionInfo.setSupportedCQDurationEncodingVersions(capabilities);
+      }
+    }
   }
 
   @Override