Fix time partition boundary overflows and splitter partition order

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/fill/filter/FixedIntervalFillFilter.java

@@ -37,11 +37,11 @@ public boolean needFill(long time, long previousTime) {
     return isTimeDistanceLessThanOrEqual(time, previousTime, timeInterval);
   }
 
-  private boolean isTimeDistanceLessThanOrEqual(long left, long right, long maxDistance) {
+  private static boolean isTimeDistanceLessThanOrEqual(long left, long right, long maxDistance) {
     if (maxDistance < 0) {
       return false;
     }
-    long distance = left >= right ? left - right : right - left;
+    final long distance = left >= right ? left - right : right - left;
     return Long.compareUnsigned(distance, maxDistance) <= 0;
   }
 }

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/fill/linear/LinearFill.java

@@ -157,9 +157,8 @@ private boolean fill(
   }
 
   private double getFactor(long currentTime) {
-    return nextTimeInCurrentColumn - previousTime == 0
-        ? 0.0
-        : ((double) (currentTime - previousTime)) / (nextTimeInCurrentColumn - previousTime);
+    double timeRange = (double) nextTimeInCurrentColumn - (double) previousTime;
+    return timeRange == 0 ? 0.0 : ((double) currentTime - (double) previousTime) / timeRange;
   }
 
   /**

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/process/gapfill/AbstractGapFillOperator.java

@@ -39,6 +39,7 @@
 import java.util.List;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.iotdb.calc.transformation.dag.column.unary.scalar.DateBinFunctionColumnTransformer.saturatingAdd;
 
 abstract class AbstractGapFillOperator implements ProcessOperator {
 
@@ -127,7 +128,9 @@ public TsBlock next() throws Exception {
       // -1 because we should not include current row, current row will be appended in
       // writeCurrentRow
       long currentEndTime =
-          timeColumn.isNull(i) ? endTime : block.getColumn(timeColumnIndex).getLong(i) - 1;
+          timeColumn.isNull(i)
+              ? endTime
+              : saturatingAdd(block.getColumn(timeColumnIndex).getLong(i), -1);
       fillGaps(block, i, currentEndTime);
       writeCurrentRow(block, i);
     }
@@ -157,8 +160,12 @@ private void writeCurrentRow(TsBlock block, int rowIndex) {
 
   private void fillGaps(TsBlock block, int rowIndex, long currentEndTime) {
     while (currentTime <= currentEndTime) {
+      long previousTime = currentTime;
       gapFillRow(currentTime, block, rowIndex);
       nextTime();
+      if (currentTime <= previousTime) {
+        break;
+      }
     }
   }
 

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/dag/column/unary/scalar/DateBinFunctionColumnTransformer.java

@@ -28,6 +28,7 @@
 import org.apache.tsfile.block.column.ColumnBuilder;
 import org.apache.tsfile.read.common.type.Type;
 
+import java.math.BigInteger;
 import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneId;
@@ -38,6 +39,8 @@
 
   private static final long NANOSECONDS_IN_MILLISECOND = 1_000_000;
   private static final long NANOSECONDS_IN_MICROSECOND = 1_000;
+  private static final BigInteger BIG_LONG_MIN = BigInteger.valueOf(Long.MIN_VALUE);
+  private static final BigInteger BIG_LONG_MAX = BigInteger.valueOf(Long.MAX_VALUE);
 
   private final int monthDuration;
   private final long nonMonthDuration;
@@ -150,12 +153,18 @@
       return convertToTimestamp(binStart, zoneId);
     }
 
-    long diff = source - origin;
-    long n = diff >= 0 ? diff / nonMonthDuration : (diff - nonMonthDuration + 1) / nonMonthDuration;
-    return origin + (n * nonMonthDuration);
+    return saturateToLong(getNonMonthDateBinStart(source, origin, nonMonthDuration));
   }
 
   public long[] dateBinStartEnd(long source) {
+    return dateBinStartEnd(source, false);
+  }
+
+  public long[] dateBinStartEndClosed(long source) {
+    return dateBinStartEnd(source, true);
+  }
+
+  private long[] dateBinStartEnd(long source, boolean closedEnd) {
     // return source if interval is 0
     if (monthDuration == 0 && nonMonthDuration == 0) {
       return new long[] {source, source};
@@ -181,17 +190,17 @@
             binStart.minusMonths(monthDuration).minusNanos(getNanoTimeStamp(nonMonthDuration));
       }
 
-      return new long[] {
-        convertToTimestamp(binStart, zoneId),
-        convertToTimestamp(binStart.plusMonths(monthDuration), zoneId)
-      };
+      long startTime = convertToTimestamp(binStart, zoneId);
+      long endTime = convertToTimestamp(binStart.plusMonths(monthDuration), zoneId);
+      return new long[] {startTime, closedEnd ? saturatingAdd(endTime, -1) : endTime};
     }
 
-    long diff = source - origin;
-    long n = diff >= 0 ? diff / nonMonthDuration : (diff - nonMonthDuration + 1) / nonMonthDuration;
-    return new long[] {
-      origin + (n * nonMonthDuration), origin + (n * nonMonthDuration) + nonMonthDuration
-    };
+    BigInteger startTime = getNonMonthDateBinStart(source, origin, nonMonthDuration);
+    BigInteger endTime = startTime.add(BigInteger.valueOf(nonMonthDuration));
+    if (closedEnd) {
+      endTime = endTime.subtract(BigInteger.ONE);
+    }
+    return new long[] {saturateToLong(startTime), saturateToLong(endTime)};
   }
 
   public static long nextDateBin(int monthDuration, ZoneId zoneId, long currentTime) {
@@ -201,7 +210,33 @@
   }
 
   public static long nextDateBin(long nonMonthDuration, long currentTime) {
-    return currentTime + nonMonthDuration;
+    return saturatingAdd(currentTime, nonMonthDuration);
+  }
+
+  public static long saturatingAdd(long left, long right) {
+    return saturateToLong(BigInteger.valueOf(left).add(BigInteger.valueOf(right)));
+  }
+
+  private static BigInteger getNonMonthDateBinStart(
+      long source, long origin, long nonMonthDuration) {
+    BigInteger diff = BigInteger.valueOf(source).subtract(BigInteger.valueOf(origin));
+    BigInteger duration = BigInteger.valueOf(nonMonthDuration);
+    BigInteger[] quotientAndRemainder = diff.divideAndRemainder(duration);
+    BigInteger n = quotientAndRemainder[0];
+    if (diff.signum() < 0 && quotientAndRemainder[1].signum() != 0) {
+      n = n.subtract(BigInteger.ONE);
+    }
+    return BigInteger.valueOf(origin).add(n.multiply(duration));
+  }
+
+  private static long saturateToLong(BigInteger value) {
+    if (value.compareTo(BIG_LONG_MAX) > 0) {
+      return Long.MAX_VALUE;
+    }
+    if (value.compareTo(BIG_LONG_MIN) < 0) {
+      return Long.MIN_VALUE;
+    }
+    return value.longValue();
   }
 
   @Override

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionWalRetentionCalculator.java

@@ -118,9 +118,15 @@ private SubscriptionRetentionBound buildTimeRetentionBound(final long retentionM
       return SubscriptionRetentionBound.retainAll();
     }
 
-    final long cutoffTimeMs = System.currentTimeMillis() - retentionMs;
+    final long cutoffTimeMs = getTimeRetentionCutoffTimeMs(System.currentTimeMillis(), retentionMs);
     final Pair<Long, Long> deletionBound =
         consensusReqReader.getDeletionBoundBeforeTimestamp(cutoffTimeMs);
     return SubscriptionRetentionBound.of(deletionBound.left, deletionBound.right);
   }
+
+  static long getTimeRetentionCutoffTimeMs(final long currentTimeMs, final long retentionMs) {
+    return retentionMs > 0 && currentTimeMs < Long.MIN_VALUE + retentionMs
+        ? Long.MIN_VALUE
+        : currentTimeMs - retentionMs;
+  }
 }

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionWalRetentionCalculatorTest.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.consensus.iot.subscription;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class SubscriptionWalRetentionCalculatorTest {
+
+  @Test
+  public void testTimeRetentionCutoffUsesNormalSubtractionWhenSafe() {
+    assertEquals(
+        700L, SubscriptionWalRetentionCalculator.getTimeRetentionCutoffTimeMs(1000L, 300L));
+  }
+
+  @Test
+  public void testTimeRetentionCutoffSaturatesOnUnderflow() {
+    assertEquals(
+        Long.MIN_VALUE,
+        SubscriptionWalRetentionCalculator.getTimeRetentionCutoffTimeMs(Long.MIN_VALUE, 1L));
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/auth/LoginLockManager.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedDeque;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.TimeUnit;
 
 public class LoginLockManager {
   private static final Logger LOGGER = LoggerFactory.getLogger(LoginLockManager.class);
@@ -156,7 +157,7 @@ public boolean checkLock(long userId, String ip) {
       UserLockInfo userIpLock = userIpLocks.get(userIpKey);
       if (userIpLock != null) {
         long now = System.currentTimeMillis();
-        long cutoffTime = now - (passwordLockTimeMinutes * 60 * 1000L);
+        long cutoffTime = getLockWindowCutoffTime(now);
         userIpLock.removeOldFailures(cutoffTime);
         if (userIpLock.getFailureCount() >= failedLoginAttempts) {
           return true;
@@ -169,7 +170,7 @@ public boolean checkLock(long userId, String ip) {
       UserLockInfo userLock = userLocks.get(userId);
       if (userLock != null) {
         long now = System.currentTimeMillis();
-        long cutoffTime = now - (passwordLockTimeMinutes * 60 * 1000L);
+        long cutoffTime = getLockWindowCutoffTime(now);
         userLock.removeOldFailures(cutoffTime);
         return userLock.getFailureCount() >= failedLoginAttemptsPerUser;
       }
@@ -200,7 +201,7 @@ public void recordFailure(long userId, String ip) {
     }
 
     long now = System.currentTimeMillis();
-    long cutoffTime = now - (passwordLockTimeMinutes * 60 * 1000L);
+    long cutoffTime = getLockWindowCutoffTime(now);
 
     // Handle user@ip failures in sliding window
     if (failedLoginAttempts != -1) {
@@ -292,7 +293,7 @@ public void unlock(long userId, String ip) {
   /** Clean up expired locks (no failures in the sliding window) */
   public void cleanExpiredLocks() {
     long now = System.currentTimeMillis();
-    long cutoffTime = now - (passwordLockTimeMinutes * 60 * 1000L);
+    long cutoffTime = getLockWindowCutoffTime(now);
 
     // Clean expired user locks
     userLocks
@@ -334,6 +335,17 @@ private String buildUserIpKey(long userId, String ip) {
     return userId + "@" + ip;
   }
 
+  private long getLockWindowCutoffTime(long currentTimeMillis) {
+    return getLockWindowCutoffTime(currentTimeMillis, passwordLockTimeMinutes);
+  }
+
+  static long getLockWindowCutoffTime(long currentTimeMillis, int passwordLockTimeMinutes) {
+    final long lockWindowMs = TimeUnit.MINUTES.toMillis(passwordLockTimeMinutes);
+    return currentTimeMillis < Long.MIN_VALUE + lockWindowMs
+        ? Long.MIN_VALUE
+        : currentTimeMillis - lockWindowMs;
+  }
+
   private void checkForPotentialAttacks(long userId, String ip) {
     // Check if IP is locked by many users
     Set<Long> usersForIp = new HashSet<>();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/exception/query/QueryTimeoutRuntimeException.java

@@ -31,8 +31,23 @@ public class QueryTimeoutRuntimeException extends IoTDBRuntimeException {
 
   public QueryTimeoutRuntimeException(long startTime, long currentTime, long timeout) {
     super(
-        String.format(QUERY_TIMEOUT_EXCEPTION_MESSAGE, startTime, startTime + timeout, currentTime),
+        String.format(
+            QUERY_TIMEOUT_EXCEPTION_MESSAGE,
+            startTime,
+            saturatingAdd(startTime, timeout),
+            currentTime),
         QUERY_TIMEOUT.getStatusCode(),
         true);
   }
+
+  private static long saturatingAdd(long left, long right) {
+    long result = left + right;
+    if (right > 0 && result < left) {
+      return Long.MAX_VALUE;
+    }
+    if (right < 0 && result > left) {
+      return Long.MIN_VALUE;
+    }
+    return result;
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/aggregate/window/processor/TumblingWindowingProcessor.java

@@ -31,6 +31,7 @@
 
 import org.apache.tsfile.utils.Pair;
 
+import java.math.BigInteger;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
@@ -80,10 +81,12 @@ public Set<TimeSeriesWindow> mayAddWindow(
             ? slidingBoundaryTime
             : windowList.get(windowList.size() - 1).getTimestamp();
 
-    if (timeStamp >= (windowList.isEmpty() ? lastTime : lastTime + slidingInterval)) {
+    if (windowList.isEmpty()
+        ? timeStamp >= lastTime
+        : isTimestampAtOrAfterWindowEnd(timeStamp, lastTime, slidingInterval)) {
       final TimeSeriesWindow window = new TimeSeriesWindow(this, null);
       // Align to the last time + k * slidingInterval, k is a natural number
-      window.setTimestamp(((timeStamp - lastTime) / slidingInterval) * slidingInterval + lastTime);
+      window.setTimestamp(alignWindowStart(timeStamp, lastTime, slidingInterval));
       windowList.add(window);
       return Collections.singleton(window);
     }
@@ -96,12 +99,12 @@ public Pair<WindowState, WindowOutput> updateAndMaySetWindowState(
     if (timeStamp < window.getTimestamp()) {
       return new Pair<>(WindowState.IGNORE_VALUE, null);
     }
-    if (timeStamp >= window.getTimestamp() + slidingInterval) {
+    if (isTimestampAtOrAfterWindowEnd(timeStamp, window.getTimestamp(), slidingInterval)) {
       return new Pair<>(
           WindowState.EMIT_AND_PURGE_WITHOUT_COMPUTE,
           new WindowOutput()
               .setTimestamp(window.getTimestamp())
-              .setProgressTime(window.getTimestamp() + slidingInterval));
+              .setProgressTime(saturatingAdd(window.getTimestamp(), slidingInterval)));
     }
     return new Pair<>(WindowState.COMPUTE, null);
   }
@@ -110,6 +113,33 @@ public Pair<WindowState, WindowOutput> updateAndMaySetWindowState(
   public WindowOutput forceOutput(final TimeSeriesWindow window) {
     return new WindowOutput()
         .setTimestamp(window.getTimestamp())
-        .setProgressTime(window.getTimestamp() + slidingInterval);
+        .setProgressTime(saturatingAdd(window.getTimestamp(), slidingInterval));
+  }
+
+  private static boolean isTimestampAtOrAfterWindowEnd(
+      final long timestamp, final long windowStart, final long interval) {
+    return windowStart <= Long.MAX_VALUE - interval && timestamp >= windowStart + interval;
+  }
+
+  private static long alignWindowStart(
+      final long timestamp, final long baseTime, final long interval) {
+    final BigInteger base = BigInteger.valueOf(baseTime);
+    final BigInteger intervalValue = BigInteger.valueOf(interval);
+    return base.add(
+            BigInteger.valueOf(timestamp)
+                .subtract(base)
+                .divide(intervalValue)
+                .multiply(intervalValue))
+        .longValueExact();
+  }
+
+  private static long saturatingAdd(final long left, final long right) {
+    if (right > 0 && left > Long.MAX_VALUE - right) {
+      return Long.MAX_VALUE;
+    }
+    if (right < 0 && left < Long.MIN_VALUE - right) {
+      return Long.MIN_VALUE;
+    }
+    return left + right;
   }
 }