Fix overflow edge cases in query utilities

iotdb-core/calc-commons/pom.xml

@@ -99,6 +99,11 @@
             <groupId>at.yawk.lz4</groupId>
             <artifactId>lz4-java</artifactId>
         </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>

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

@@ -34,6 +34,14 @@ public FixedIntervalFillFilter(long timeInterval) {
   public boolean needFill(long time, long previousTime) {
     // the reason that we use Math.abs is that we may use order by time desc which will cause
     // previousTime is larger than time
-    return Math.abs(time - previousTime) <= timeInterval;
+    return isTimeDistanceLessThanOrEqual(time, previousTime, timeInterval);
+  }
+
+  private boolean isTimeDistanceLessThanOrEqual(long left, long right, long maxDistance) {
+    if (maxDistance < 0) {
+      return false;
+    }
+    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/source/relational/aggregation/ExtremeAccumulator.java

@@ -247,13 +247,9 @@ private void addIntInput(Column column, AggregationMask mask) {
   }
 
   private void updateIntResult(int val) {
-    int absExtVal = Math.abs(val);
     int candidateResult = extremeResult.getInt();
-    int absCandidateResult = Math.abs(extremeResult.getInt());
 
-    if (!initResult
-        || (absExtVal > absCandidateResult)
-        || (absExtVal == absCandidateResult) && val > candidateResult) {
+    if (!initResult || compareExtreme(val, candidateResult) > 0) {
       initResult = true;
       extremeResult.setInt(val);
     }
@@ -281,13 +277,9 @@ private void addLongInput(Column column, AggregationMask mask) {
   }
 
   private void updateLongResult(long val) {
-    long absExtVal = Math.abs(val);
     long candidateResult = extremeResult.getLong();
-    long absCandidateResult = Math.abs(extremeResult.getLong());
 
-    if (!initResult
-        || (absExtVal > absCandidateResult)
-        || (absExtVal == absCandidateResult) && val > candidateResult) {
+    if (!initResult || compareExtreme(val, candidateResult) > 0) {
       initResult = true;
       extremeResult.setLong(val);
     }
@@ -360,4 +352,24 @@ private void updateDoubleResult(double val) {
       extremeResult.setDouble(val);
     }
   }
+
+  private int compareExtreme(int left, int right) {
+    int absComparison = Long.compare(Math.abs((long) left), Math.abs((long) right));
+    return absComparison == 0 ? Integer.compare(left, right) : absComparison;
+  }
+
+  private int compareExtreme(long left, long right) {
+    int absComparison = compareAbs(left, right);
+    return absComparison == 0 ? Long.compare(left, right) : absComparison;
+  }
+
+  private int compareAbs(long left, long right) {
+    if (left == Long.MIN_VALUE) {
+      return right == Long.MIN_VALUE ? 0 : 1;
+    }
+    if (right == Long.MIN_VALUE) {
+      return -1;
+    }
+    return Long.compare(Math.abs(left), Math.abs(right));
+  }
 }

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedExtremeAccumulator.java

@@ -173,16 +173,16 @@ public void addIntermediate(int[] groupIds, Column argument) {
 
       switch (seriesDataType) {
         case INT32:
-          updateIntValue(groupIds[i], Math.abs(argument.getInt(i)));
+          updateIntValue(groupIds[i], argument.getInt(i));
           break;
         case INT64:
-          updateLongValue(groupIds[i], Math.abs(argument.getLong(i)));
+          updateLongValue(groupIds[i], argument.getLong(i));
           break;
         case FLOAT:
-          updateFloatValue(groupIds[i], Math.abs(argument.getFloat(i)));
+          updateFloatValue(groupIds[i], argument.getFloat(i));
           break;
         case DOUBLE:
-          updateDoubleValue(groupIds[i], Math.abs(argument.getDouble(i)));
+          updateDoubleValue(groupIds[i], argument.getDouble(i));
           break;
         case TEXT:
         case STRING:
@@ -300,7 +300,7 @@ private void addIntInput(int[] groupIds, Column valueColumn, AggregationMask mas
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
         if (!valueColumn.isNull(i)) {
-          updateIntValue(groupIds[i], Math.abs(valueColumn.getInt(i)));
+          updateIntValue(groupIds[i], valueColumn.getInt(i));
         }
       }
     } else {
@@ -309,15 +309,15 @@ private void addIntInput(int[] groupIds, Column valueColumn, AggregationMask mas
       for (int i = 0; i < positionCount; i++) {
         position = selectedPositions[i];
         if (!valueColumn.isNull(position)) {
-          updateIntValue(groupIds[position], Math.abs(valueColumn.getInt(position)));
+          updateIntValue(groupIds[position], valueColumn.getInt(position));
         }
       }
     }
   }
 
   protected void updateIntValue(int groupId, int value) {
-    int max = intValues.get(groupId);
-    if (value >= max) {
+    int candidate = intValues.get(groupId);
+    if (!inits.get(groupId) || compareExtreme(value, candidate) >= 0) {
       inits.set(groupId, true);
       intValues.set(groupId, value);
     }
@@ -329,7 +329,7 @@ private void addLongInput(int[] groupIds, Column valueColumn, AggregationMask ma
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
         if (!valueColumn.isNull(i)) {
-          updateLongValue(groupIds[i], Math.abs(valueColumn.getLong(i)));
+          updateLongValue(groupIds[i], valueColumn.getLong(i));
         }
       }
     } else {
@@ -338,15 +338,15 @@ private void addLongInput(int[] groupIds, Column valueColumn, AggregationMask ma
       for (int i = 0; i < positionCount; i++) {
         position = selectedPositions[i];
         if (!valueColumn.isNull(position)) {
-          updateLongValue(groupIds[position], Math.abs(valueColumn.getLong(position)));
+          updateLongValue(groupIds[position], valueColumn.getLong(position));
         }
       }
     }
   }
 
   protected void updateLongValue(int groupId, long value) {
-    long max = longValues.get(groupId);
-    if (value >= max) {
+    long candidate = longValues.get(groupId);
+    if (!inits.get(groupId) || compareExtreme(value, candidate) >= 0) {
       inits.set(groupId, true);
       longValues.set(groupId, value);
     }
@@ -358,7 +358,7 @@ private void addFloatInput(int[] groupIds, Column valueColumn, AggregationMask m
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
         if (!valueColumn.isNull(i)) {
-          updateFloatValue(groupIds[i], Math.abs(valueColumn.getFloat(i)));
+          updateFloatValue(groupIds[i], valueColumn.getFloat(i));
         }
       }
     } else {
@@ -367,15 +367,15 @@ private void addFloatInput(int[] groupIds, Column valueColumn, AggregationMask m
       for (int i = 0; i < positionCount; i++) {
         position = selectedPositions[i];
         if (!valueColumn.isNull(position)) {
-          updateFloatValue(groupIds[position], Math.abs(valueColumn.getFloat(position)));
+          updateFloatValue(groupIds[position], valueColumn.getFloat(position));
         }
       }
     }
   }
 
   protected void updateFloatValue(int groupId, float value) {
-    float max = floatValues.get(groupId);
-    if (value >= max) {
+    float candidate = floatValues.get(groupId);
+    if (!inits.get(groupId) || compareExtreme(value, candidate) >= 0) {
       inits.set(groupId, true);
       floatValues.set(groupId, value);
     }
@@ -387,7 +387,7 @@ private void addDoubleInput(int[] groupIds, Column valueColumn, AggregationMask
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
         if (!valueColumn.isNull(i)) {
-          updateDoubleValue(groupIds[i], Math.abs(valueColumn.getDouble(i)));
+          updateDoubleValue(groupIds[i], valueColumn.getDouble(i));
         }
       }
     } else {
@@ -396,17 +396,47 @@ private void addDoubleInput(int[] groupIds, Column valueColumn, AggregationMask
       for (int i = 0; i < positionCount; i++) {
         position = selectedPositions[i];
         if (!valueColumn.isNull(position)) {
-          updateDoubleValue(groupIds[position], Math.abs(valueColumn.getDouble(position)));
+          updateDoubleValue(groupIds[position], valueColumn.getDouble(position));
         }
       }
     }
   }
 
   protected void updateDoubleValue(int groupId, double value) {
-    double max = doubleValues.get(groupId);
-    if (value >= max) {
+    double candidate = doubleValues.get(groupId);
+    if (!inits.get(groupId) || compareExtreme(value, candidate) >= 0) {
       inits.set(groupId, true);
       doubleValues.set(groupId, value);
     }
   }
+
+  private int compareExtreme(int left, int right) {
+    int absComparison = Long.compare(Math.abs((long) left), Math.abs((long) right));
+    return absComparison == 0 ? Integer.compare(left, right) : absComparison;
+  }
+
+  private int compareExtreme(long left, long right) {
+    int absComparison = compareAbs(left, right);
+    return absComparison == 0 ? Long.compare(left, right) : absComparison;
+  }
+
+  private int compareAbs(long left, long right) {
+    if (left == Long.MIN_VALUE) {
+      return right == Long.MIN_VALUE ? 0 : 1;
+    }
+    if (right == Long.MIN_VALUE) {
+      return -1;
+    }
+    return Long.compare(Math.abs(left), Math.abs(right));
+  }
+
+  private int compareExtreme(float left, float right) {
+    int absComparison = Float.compare(Math.abs(left), Math.abs(right));
+    return absComparison == 0 ? Float.compare(left, right) : absComparison;
+  }
+
+  private int compareExtreme(double left, double right) {
+    int absComparison = Double.compare(Math.abs(left), Math.abs(right));
+    return absComparison == 0 ? Double.compare(left, right) : absComparison;
+  }
 }

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

@@ -0,0 +1,41 @@
+/*
+ * 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.calc.execution.operator.process.fill.filter;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class FixedIntervalFillFilterTest {
+
+  @Test
+  public void needFillHandlesOverflowedTimeDistance() {
+    FixedIntervalFillFilter filter = new FixedIntervalFillFilter(1);
+
+    Assert.assertTrue(filter.needFill(Long.MIN_VALUE, Long.MIN_VALUE + 1));
+    Assert.assertFalse(filter.needFill(Long.MIN_VALUE, Long.MAX_VALUE));
+  }
+
+  @Test
+  public void needFillRejectsNegativeInterval() {
+    FixedIntervalFillFilter filter = new FixedIntervalFillFilter(-1);
+
+    Assert.assertFalse(filter.needFill(0, 0));
+  }
+}