[To dev/1.3] Harden timestamp arithmetic and boundary handling across IoTDB

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

@@ -19,6 +19,8 @@
 
 package org.apache.iotdb.db.exception.query;
 
+import com.google.common.math.LongMath;
+
 /** This class is used to throw run time exception when query is time out. */
 public class QueryTimeoutRuntimeException extends RuntimeException {
   public static final String QUERY_TIMEOUT_EXCEPTION_MESSAGE =
@@ -27,7 +29,10 @@ public class QueryTimeoutRuntimeException extends RuntimeException {
   public QueryTimeoutRuntimeException(long startTime, long currentTime, long timeout) {
     super(
         String.format(
-            QUERY_TIMEOUT_EXCEPTION_MESSAGE, startTime, startTime + timeout, currentTime));
+            QUERY_TIMEOUT_EXCEPTION_MESSAGE,
+            startTime,
+            LongMath.saturatedAdd(startTime, timeout),
+            currentTime));
   }
 
   public QueryTimeoutRuntimeException(String message) {

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

@@ -30,11 +30,13 @@
 
 import org.apache.tsfile.utils.Pair;
 
+import java.math.BigInteger;
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import static com.google.common.math.LongMath.saturatedAdd;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_SLIDING_BOUNDARY_TIME_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_SLIDING_BOUNDARY_TIME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_SLIDING_SECONDS_DEFAULT_VALUE;
@@ -78,10 +80,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);
     }
@@ -94,12 +98,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(saturatedAdd(window.getTimestamp(), slidingInterval)));
     }
     return new Pair<>(WindowState.COMPUTE, null);
   }
@@ -108,6 +112,23 @@ public Pair<WindowState, WindowOutput> updateAndMaySetWindowState(
   public WindowOutput forceOutput(final TimeSeriesWindow window) {
     return new WindowOutput()
         .setTimestamp(window.getTimestamp())
-        .setProgressTime(window.getTimestamp() + slidingInterval);
+        .setProgressTime(saturatedAdd(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();
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/DownSamplingTimeUtils.java

@@ -0,0 +1,43 @@
+/*
+ * 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.db.pipe.processor.downsampling;
+
+public class DownSamplingTimeUtils {
+
+  private DownSamplingTimeUtils() {
+    // Utility class.
+  }
+
+  public static boolean isTimeDistanceLessThanOrEqualTo(long left, long right, long distance) {
+    if (distance < 0) {
+      return false;
+    }
+    final long difference = left >= right ? left - right : right - left;
+    return Long.compareUnsigned(difference, distance) <= 0;
+  }
+
+  public static boolean isTimeDistanceGreaterThanOrEqualTo(long left, long right, long distance) {
+    if (distance < 0) {
+      return true;
+    }
+    final long difference = left >= right ? left - right : right - left;
+    return Long.compareUnsigned(difference, distance) >= 0;
+  }
+}

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/changing/ChangingValueFilter.java

@@ -24,6 +24,9 @@
 import java.time.LocalDate;
 import java.util.Objects;
 
+import static org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingTimeUtils.isTimeDistanceGreaterThanOrEqualTo;
+import static org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingTimeUtils.isTimeDistanceLessThanOrEqualTo;
+
 public class ChangingValueFilter<T> {
 
   private final ChangingValueSamplingProcessor processor;
@@ -59,13 +62,13 @@ public boolean filter(final long timestamp, final T value) {
   }
 
   private boolean tryFilter(final long timestamp, final T value) {
-    final long timeDiff = Math.abs(timestamp - lastStoredTimestamp);
-
-    if (timeDiff <= processor.getCompressionMinTimeInterval()) {
+    if (isTimeDistanceLessThanOrEqualTo(
+        timestamp, lastStoredTimestamp, processor.getCompressionMinTimeInterval())) {
       return false;
     }
 
-    if (timeDiff >= processor.getCompressionMaxTimeInterval()) {
+    if (isTimeDistanceGreaterThanOrEqualTo(
+        timestamp, lastStoredTimestamp, processor.getCompressionMaxTimeInterval())) {
       reset(timestamp, value);
       return true;
     }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/sdt/SwingingDoorTrendingFilter.java

@@ -24,6 +24,10 @@
 import java.time.LocalDate;
 import java.util.Objects;
 
+import static org.apache.iotdb.commons.utils.CommonDateTimeUtils.timeDifferenceAsDouble;
+import static org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingTimeUtils.isTimeDistanceGreaterThanOrEqualTo;
+import static org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingTimeUtils.isTimeDistanceLessThanOrEqualTo;
+
 public class SwingingDoorTrendingFilter<T> {
 
   private final SwingingDoorTrendingSamplingProcessor processor;
@@ -85,14 +89,13 @@ public boolean filter(final long timestamp, final T value) {
   }
 
   private boolean tryFilter(final long timestamp, final T value) {
-    final long timeDiff = timestamp - lastStoredTimestamp;
-    final long absTimeDiff = Math.abs(timeDiff);
-
-    if (absTimeDiff <= processor.getCompressionMinTimeInterval()) {
+    if (isTimeDistanceLessThanOrEqualTo(
+        timestamp, lastStoredTimestamp, processor.getCompressionMinTimeInterval())) {
       return false;
     }
 
-    if (absTimeDiff >= processor.getCompressionMaxTimeInterval()) {
+    if (isTimeDistanceGreaterThanOrEqualTo(
+        timestamp, lastStoredTimestamp, processor.getCompressionMaxTimeInterval())) {
       reset(timestamp, value);
       return true;
     }
@@ -114,6 +117,7 @@ private boolean tryFilter(final long timestamp, final T value) {
     final double doubleValue = Double.parseDouble(value.toString());
     final double lastStoredDoubleValue = Double.parseDouble(lastStoredValue.toString());
     final double valueDiff = doubleValue - lastStoredDoubleValue;
+    final double timeDiff = timeDifferenceAsDouble(timestamp, lastStoredTimestamp);
 
     final double currentUpperSlope = (valueDiff - processor.getCompressionDeviation()) / timeDiff;
     if (currentUpperSlope > upperDoor) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/processor/downsampling/tumbling/TumblingTimeSamplingProcessor.java

@@ -39,6 +39,7 @@
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_DOWN_SAMPLING_SPLIT_FILE_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TUMBLING_TIME_INTERVAL_SECONDS_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant.PROCESSOR_TUMBLING_TIME_INTERVAL_SECONDS_KEY;
+import static org.apache.iotdb.db.pipe.processor.downsampling.DownSamplingTimeUtils.isTimeDistanceGreaterThanOrEqualTo;
 
 public class TumblingTimeSamplingProcessor extends DownSamplingProcessor {
 
@@ -113,7 +114,8 @@ protected void processRow(
       final Long lastSampleTime = pathLastObjectCache.getPartialPathLastObject(timeSeriesSuffix);
 
       if (lastSampleTime == null
-          || Math.abs(currentRowTime - lastSampleTime) >= intervalInCurrentPrecision) {
+          || isTimeDistanceGreaterThanOrEqualTo(
+              currentRowTime, lastSampleTime, intervalInCurrentPrecision)) {
         try {
           rowCollector.collectRow(row);
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/DataRegionWatermarkInjector.java

@@ -24,6 +24,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static com.google.common.math.LongMath.saturatedAdd;
+
 public class DataRegionWatermarkInjector {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DataRegionWatermarkInjector.class);
@@ -66,7 +68,11 @@ public PipeWatermarkEvent inject() {
   }
 
   private static long calculateNextInjectionTime(long injectionIntervalInMs) {
-    final long currentTime = System.currentTimeMillis();
-    return currentTime / injectionIntervalInMs * injectionIntervalInMs + injectionIntervalInMs;
+    return calculateNextInjectionTime(System.currentTimeMillis(), injectionIntervalInMs);
+  }
+
+  static long calculateNextInjectionTime(long currentTime, long injectionIntervalInMs) {
+    return saturatedAdd(
+        currentTime / injectionIntervalInMs * injectionIntervalInMs, injectionIntervalInMs);
   }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java

@@ -231,11 +231,11 @@ public void customize(
     }
 
     startTimePartitionIdLowerBound =
-        (realtimeDataExtractionStartTime % TimePartitionUtils.getTimePartitionInterval() == 0)
+        TimePartitionUtils.isTimePartitionStartTime(realtimeDataExtractionStartTime)
             ? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionStartTime)
             : TimePartitionUtils.getTimePartitionId(realtimeDataExtractionStartTime) + 1;
     endTimePartitionIdUpperBound =
-        (realtimeDataExtractionEndTime % TimePartitionUtils.getTimePartitionInterval() == 0)
+        TimePartitionUtils.isTimePartitionStartTime(realtimeDataExtractionEndTime)
             ? TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime)
             : TimePartitionUtils.getTimePartitionId(realtimeDataExtractionEndTime) - 1;
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/TimeDurationAccumulator.java

@@ -87,7 +87,7 @@ public void outputFinal(ColumnBuilder tsBlockBuilder) {
     if (!initResult) {
       tsBlockBuilder.appendNull();
     } else {
-      tsBlockBuilder.writeLong(maxTime - minTime);
+      tsBlockBuilder.writeLong(saturatingTimeDifference(maxTime, minTime));
     }
   }
 
@@ -127,4 +127,9 @@ protected void updateMinTime(long curTime) {
     initResult = true;
     minTime = Math.min(minTime, curTime);
   }
+
+  private static long saturatingTimeDifference(long maxTime, long minTime) {
+    long timeDifference = maxTime - minTime;
+    return timeDifference < 0 ? Long.MAX_VALUE : timeDifference;
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/timerangeiterator/AggrWindowIterator.java

@@ -27,6 +27,8 @@
 
 import java.time.ZoneId;
 
+import static com.google.common.math.LongMath.saturatedAdd;
+
 /**
  * This class iteratively generates aggregated time windows.
  *
@@ -86,23 +88,22 @@ private TimeRange getLeftmostTimeRange() {
       retEndTime =
           Math.min(DateTimeUtils.calcPositiveIntervalByMonth(startTime, interval, zoneId), endTime);
     } else {
-      retEndTime = Math.min(startTime + interval.nonMonthDuration, endTime);
+      retEndTime = Math.min(saturatedAdd(startTime, interval.nonMonthDuration), endTime);
     }
     return new TimeRange(startTime, retEndTime);
   }
 
   private TimeRange getRightmostTimeRange() {
     long retStartTime;
     long retEndTime;
-    long queryRange = endTime - startTime;
     long intervalNum;
 
     if (slidingStep.containsMonth()) {
       intervalNum =
-          (long)
-              Math.ceil(
-                  (double) queryRange
-                      / (slidingStep.getMaxTotalDuration(TimestampPrecisionUtils.currPrecision)));
+          ITimeRangeIterator.ceilDivTimeRange(
+              startTime,
+              endTime,
+              slidingStep.getMaxTotalDuration(TimestampPrecisionUtils.currPrecision));
       long tempRetStartTime =
           DateTimeUtils.calcPositiveIntervalByMonth(
               startTime, slidingStep.multiple(intervalNum - 1), zoneId);
@@ -116,8 +117,11 @@ private TimeRange getRightmostTimeRange() {
       }
       intervalNum -= 1;
     } else {
-      intervalNum = (long) Math.ceil(queryRange / (double) slidingStep.nonMonthDuration);
-      retStartTime = slidingStep.nonMonthDuration * (intervalNum - 1) + startTime;
+      intervalNum =
+          ITimeRangeIterator.ceilDivTimeRange(startTime, endTime, slidingStep.nonMonthDuration);
+      retStartTime =
+          ITimeRangeIterator.rightmostTimeRangeStart(
+              startTime, endTime, slidingStep.nonMonthDuration);
     }
 
     if (interval.containsMonth()) {
@@ -129,7 +133,7 @@ private TimeRange getRightmostTimeRange() {
                   startTime, interval.merge(slidingStep.multiple(intervalNum - 1)), zoneId),
               endTime);
     } else {
-      retEndTime = Math.min(retStartTime + interval.nonMonthDuration, endTime);
+      retEndTime = Math.min(saturatedAdd(retStartTime, interval.nonMonthDuration), endTime);
     }
     return new TimeRange(retStartTime, retEndTime);
   }
@@ -155,6 +159,10 @@ public boolean hasNextTimeRange() {
             DateTimeUtils.calcPositiveIntervalByMonth(
                 startTime, slidingStep.multiple(timeRangeCount), zoneId);
       } else {
+        if (!ITimeRangeIterator.canMoveForward(
+            curStartTime, slidingStep.nonMonthDuration, endTime)) {
+          return false;
+        }
         retStartTime = curStartTime + slidingStep.nonMonthDuration;
       }
       // This is an open interval , [0-100)
@@ -167,6 +175,10 @@ public boolean hasNextTimeRange() {
         throw new UnsupportedOperationException(
             "Ascending is not supported when sliding step contains month.");
       } else {
+        if (!ITimeRangeIterator.canMoveBackward(
+            curStartTime, slidingStep.nonMonthDuration, startTime)) {
+          return false;
+        }
         retStartTime = curStartTime - slidingStep.nonMonthDuration;
       }
       if (retStartTime < startTime) {
@@ -179,7 +191,7 @@ public boolean hasNextTimeRange() {
           DateTimeUtils.calcPositiveIntervalByMonth(
               startTime, slidingStep.multiple(timeRangeCount).merge(interval), zoneId);
     } else {
-      retEndTime = retStartTime + interval.nonMonthDuration;
+      retEndTime = saturatedAdd(retStartTime, interval.nonMonthDuration);
     }
     retEndTime = Math.min(retEndTime, endTime);
     curTimeRange = new TimeRange(retStartTime, retEndTime);
@@ -209,15 +221,14 @@ public long currentOutputTime() {
 
   @Override
   public long getTotalIntervalNum() {
-    long queryRange = endTime - startTime;
     long intervalNum;
 
     if (slidingStep.containsMonth()) {
       intervalNum =
-          (long)
-              Math.ceil(
-                  (double) queryRange
-                      / (slidingStep.getMaxTotalDuration(TimestampPrecisionUtils.currPrecision)));
+          ITimeRangeIterator.ceilDivTimeRange(
+              startTime,
+              endTime,
+              slidingStep.getMaxTotalDuration(TimestampPrecisionUtils.currPrecision));
       long retStartTime =
           DateTimeUtils.calcPositiveIntervalByMonth(
               startTime, slidingStep.multiple(intervalNum), zoneId);
@@ -228,7 +239,8 @@ public long getTotalIntervalNum() {
                 startTime, slidingStep.multiple(intervalNum), zoneId);
       }
     } else {
-      intervalNum = (long) Math.ceil(queryRange / (double) slidingStep.nonMonthDuration);
+      intervalNum =
+          ITimeRangeIterator.ceilDivTimeRange(startTime, endTime, slidingStep.nonMonthDuration);
     }
     return intervalNum;
   }