Skip to content

Commit 2c0c307

Browse files
CaideyipiCaideyipi
authored
Harden timestamp arithmetic and boundary handling across IoTDB (#17893)
1 parent fd39898 commit 2c0c307

105 files changed

Lines changed: 2849 additions & 570 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,11 +37,11 @@ public boolean needFill(long time, long previousTime) {
3737
return isTimeDistanceLessThanOrEqual(time, previousTime, timeInterval);
3838
}
3939

40-
private boolean isTimeDistanceLessThanOrEqual(long left, long right, long maxDistance) {
40+
private static boolean isTimeDistanceLessThanOrEqual(long left, long right, long maxDistance) {
4141
if (maxDistance < 0) {
4242
return false;
4343
}
44-
long distance = left >= right ? left - right : right - left;
44+
final long distance = left >= right ? left - right : right - left;
4545
return Long.compareUnsigned(distance, maxDistance) <= 0;
4646
}
4747
}

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import java.util.Optional;
2929

3030
import static com.google.common.base.Preconditions.checkArgument;
31+
import static org.apache.iotdb.commons.utils.CommonDateTimeUtils.timeDifferenceAsDouble;
3132

3233
/**
3334
* The result of Linear Fill functions at timestamp "T" is calculated by performing a linear fitting
@@ -157,9 +158,8 @@ private boolean fill(
157158
}
158159

159160
private double getFactor(long currentTime) {
160-
return nextTimeInCurrentColumn - previousTime == 0
161-
? 0.0
162-
: ((double) (currentTime - previousTime)) / (nextTimeInCurrentColumn - previousTime);
161+
double timeRange = timeDifferenceAsDouble(nextTimeInCurrentColumn, previousTime);
162+
return timeRange == 0 ? 0.0 : timeDifferenceAsDouble(currentTime, previousTime) / timeRange;
163163
}
164164

165165
/**

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

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -124,11 +124,12 @@ public TsBlock next() throws Exception {
124124
}
125125

126126
Column timeColumn = block.getColumn(timeColumnIndex);
127-
// -1 because we should not include current row, current row will be appended in
128-
// writeCurrentRow
129-
long currentEndTime =
130-
timeColumn.isNull(i) ? endTime : block.getColumn(timeColumnIndex).getLong(i) - 1;
131-
fillGaps(block, i, currentEndTime);
127+
if (timeColumn.isNull(i)) {
128+
fillGaps(block, i, endTime, true);
129+
} else {
130+
// The real row is appended by writeCurrentRow, so its timestamp is an exclusive bound.
131+
fillGaps(block, i, timeColumn.getLong(i), false);
132+
}
132133
writeCurrentRow(block, i);
133134
}
134135
lastGroupKey = new SortKey(block, size - 1);
@@ -156,9 +157,18 @@ private void writeCurrentRow(TsBlock block, int rowIndex) {
156157
}
157158

158159
private void fillGaps(TsBlock block, int rowIndex, long currentEndTime) {
159-
while (currentTime <= currentEndTime) {
160+
fillGaps(block, rowIndex, currentEndTime, true);
161+
}
162+
163+
private void fillGaps(
164+
TsBlock block, int rowIndex, long currentEndTime, boolean endTimeInclusive) {
165+
while (currentTime < currentEndTime || (endTimeInclusive && currentTime == currentEndTime)) {
166+
long previousTime = currentTime;
160167
gapFillRow(currentTime, block, rowIndex);
161168
nextTime();
169+
if (currentTime <= previousTime) {
170+
break;
171+
}
162172
}
163173
}
164174

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

Lines changed: 53 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -28,12 +28,16 @@
2828
import org.apache.tsfile.block.column.ColumnBuilder;
2929
import org.apache.tsfile.read.common.type.Type;
3030

31+
import java.math.BigInteger;
3132
import java.time.Instant;
3233
import java.time.LocalDateTime;
3334
import java.time.ZoneId;
3435
import java.time.temporal.ChronoUnit;
3536
import java.util.concurrent.TimeUnit;
3637

38+
import static com.google.common.math.LongMath.saturatedAdd;
39+
import static org.apache.iotdb.commons.utils.CommonDateTimeUtils.saturateToLong;
40+
3741
public class DateBinFunctionColumnTransformer extends UnaryColumnTransformer {
3842

3943
private static final long NANOSECONDS_IN_MILLISECOND = 1_000_000;
@@ -150,12 +154,23 @@ public static long dateBin(
150154
return convertToTimestamp(binStart, zoneId);
151155
}
152156

153-
long diff = source - origin;
154-
long n = diff >= 0 ? diff / nonMonthDuration : (diff - nonMonthDuration + 1) / nonMonthDuration;
155-
return origin + (n * nonMonthDuration);
157+
try {
158+
return getNonMonthDateBinStart(source, origin, nonMonthDuration);
159+
} catch (ArithmeticException e) {
160+
return saturateToLong(
161+
getNonMonthDateBinStartWithBigInteger(source, origin, nonMonthDuration));
162+
}
156163
}
157164

158165
public long[] dateBinStartEnd(long source) {
166+
return dateBinStartEnd(source, false);
167+
}
168+
169+
public long[] dateBinStartEndClosed(long source) {
170+
return dateBinStartEnd(source, true);
171+
}
172+
173+
private long[] dateBinStartEnd(long source, boolean closedEnd) {
159174
// return source if interval is 0
160175
if (monthDuration == 0 && nonMonthDuration == 0) {
161176
return new long[] {source, source};
@@ -181,17 +196,24 @@ public long[] dateBinStartEnd(long source) {
181196
binStart.minusMonths(monthDuration).minusNanos(getNanoTimeStamp(nonMonthDuration));
182197
}
183198

184-
return new long[] {
185-
convertToTimestamp(binStart, zoneId),
186-
convertToTimestamp(binStart.plusMonths(monthDuration), zoneId)
187-
};
199+
long startTime = convertToTimestamp(binStart, zoneId);
200+
long endTime = convertToTimestamp(binStart.plusMonths(monthDuration), zoneId);
201+
return new long[] {startTime, closedEnd ? saturatedAdd(endTime, -1) : endTime};
188202
}
189203

190-
long diff = source - origin;
191-
long n = diff >= 0 ? diff / nonMonthDuration : (diff - nonMonthDuration + 1) / nonMonthDuration;
192-
return new long[] {
193-
origin + (n * nonMonthDuration), origin + (n * nonMonthDuration) + nonMonthDuration
194-
};
204+
try {
205+
long startTime = getNonMonthDateBinStart(source, origin, nonMonthDuration);
206+
long endOffset = closedEnd ? nonMonthDuration - 1 : nonMonthDuration;
207+
return new long[] {startTime, saturatedAdd(startTime, endOffset)};
208+
} catch (ArithmeticException e) {
209+
BigInteger startTime =
210+
getNonMonthDateBinStartWithBigInteger(source, origin, nonMonthDuration);
211+
BigInteger endTime = startTime.add(BigInteger.valueOf(nonMonthDuration));
212+
if (closedEnd) {
213+
endTime = endTime.subtract(BigInteger.ONE);
214+
}
215+
return new long[] {saturateToLong(startTime), saturateToLong(endTime)};
216+
}
195217
}
196218

197219
public static long nextDateBin(int monthDuration, ZoneId zoneId, long currentTime) {
@@ -201,7 +223,25 @@ public static long nextDateBin(int monthDuration, ZoneId zoneId, long currentTim
201223
}
202224

203225
public static long nextDateBin(long nonMonthDuration, long currentTime) {
204-
return currentTime + nonMonthDuration;
226+
return saturatedAdd(currentTime, nonMonthDuration);
227+
}
228+
229+
private static long getNonMonthDateBinStart(long source, long origin, long nonMonthDuration) {
230+
long diff = Math.subtractExact(source, origin);
231+
long stepCount = Math.floorDiv(diff, nonMonthDuration);
232+
return Math.addExact(origin, Math.multiplyExact(stepCount, nonMonthDuration));
233+
}
234+
235+
private static BigInteger getNonMonthDateBinStartWithBigInteger(
236+
long source, long origin, long nonMonthDuration) {
237+
BigInteger diff = BigInteger.valueOf(source).subtract(BigInteger.valueOf(origin));
238+
BigInteger duration = BigInteger.valueOf(nonMonthDuration);
239+
BigInteger[] quotientAndRemainder = diff.divideAndRemainder(duration);
240+
BigInteger n = quotientAndRemainder[0];
241+
if (diff.signum() < 0 && quotientAndRemainder[1].signum() != 0) {
242+
n = n.subtract(BigInteger.ONE);
243+
}
244+
return BigInteger.valueOf(origin).add(n.multiply(duration));
205245
}
206246

207247
@Override

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

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,9 +128,15 @@ private SubscriptionRetentionBound buildTimeRetentionBound(final long retentionM
128128
return SubscriptionRetentionBound.retainAll();
129129
}
130130

131-
final long cutoffTimeMs = System.currentTimeMillis() - retentionMs;
131+
final long cutoffTimeMs = getTimeRetentionCutoffTimeMs(System.currentTimeMillis(), retentionMs);
132132
final Pair<Long, Long> deletionBound =
133133
consensusReqReader.getDeletionBoundBeforeTimestamp(cutoffTimeMs);
134134
return SubscriptionRetentionBound.of(deletionBound.left, deletionBound.right);
135135
}
136+
137+
static long getTimeRetentionCutoffTimeMs(final long currentTimeMs, final long retentionMs) {
138+
return retentionMs > 0 && currentTimeMs < Long.MIN_VALUE + retentionMs
139+
? Long.MIN_VALUE
140+
: currentTimeMs - retentionMs;
141+
}
136142
}

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

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,19 @@
3333

3434
public class SubscriptionWalRetentionCalculatorTest {
3535

36+
@Test
37+
public void testTimeRetentionCutoffUsesNormalSubtractionWhenSafe() {
38+
assertEquals(
39+
700L, SubscriptionWalRetentionCalculator.getTimeRetentionCutoffTimeMs(1000L, 300L));
40+
}
41+
42+
@Test
43+
public void testTimeRetentionCutoffSaturatesOnUnderflow() {
44+
assertEquals(
45+
Long.MIN_VALUE,
46+
SubscriptionWalRetentionCalculator.getTimeRetentionCutoffTimeMs(Long.MIN_VALUE, 1L));
47+
}
48+
3649
@Test
3750
public void testCommittedProgressKeepsMoreWalThanTopicRetention() {
3851
final SubscriptionWalRetentionCalculator calculator =

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,18 @@ public final class DataNodeQueryMessages {
2929
"This node isn't instance of SchemaEntityNode.";
3030
public static final String THIS_NODE_ISN_T_INSTANCE_OF_SCHEMAMEASUREMENTNODE =
3131
"This node isn't instance of SchemaMeasurementNode.";
32+
public static final String EXCEPTION_TIME_DURATION_IS_OUT_OF_RANGE_A56B1A29 =
33+
"Time duration is out of range.";
34+
public static final String EXCEPTION_GENERATED_TIME_COLUMN_IS_OUT_OF_RANGE_43AB0C2A =
35+
"Generated time column is out of range.";
36+
public static final String
37+
EXCEPTION_THE_TIME_PREDICATE_DOES_NOT_SELECT_ANY_TIME_RANGE_ARG_98DBCE32 =
38+
"The time predicate does not select any time range: %s";
39+
public static final String
40+
EXCEPTION_RIGHT_CLOSED_GROUP_BY_TIME_WITH_LONG_MAX_VALUE_END_TIME_CANNOT_BE_REPRESENTED_AS_A_SINGLE_RIGHT_OPEN_TIME_FILTER_772F893F =
41+
"Right-closed GROUP BY TIME with Long.MAX_VALUE end time cannot be represented as a single right-open time filter.";
42+
public static final String EXCEPTION_DATE_EXPRESSION_IS_OUT_OF_RANGE_ARG_ED35A8A1 =
43+
"Date expression is out of range: %s";
3244

3345
// --- Execution / Aggregation ---
3446

‎iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ private StorageEngineMessages() {}
4747
public static final String REBOOT_ALL_TIMED_SERVICE_SUCCESSFULLY = "Reboot all timed service successfully";
4848
public static final String FAILED_TO_DELETE = "Failed to delete: {} -> {}";
4949
public static final String FAILED_TO_CHECK_OBJECT_FILES = "Failed to check Object Files: {}";
50+
public static final String EXCEPTION_CANNOT_REPAIR_ARG_BECAUSE_TARGET_FILE_ALREADY_EXISTS_AND_TIMESTAMP_IS_LONG_MAX_VALUE_F29F630A =
51+
"Cannot repair %s because the target file already exists and the file timestamp is Long.MAX_VALUE";
5052

5153
// ======================== Buffer Cache ========================
5254

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,19 @@ public final class DataNodeQueryMessages {
3838
"未知的数据类型：";
3939
public static final String COUNT_IF_WITH_SLIDINGWINDOW_IS_NOT_SUPPORTED_NOW =
4040
"目前不支持 COUNT_IF 与滑动窗口组合使用";
41+
public static final String EXCEPTION_TIME_DURATION_IS_OUT_OF_RANGE_A56B1A29 =
42+
"时间间隔超出范围。";
43+
public static final String EXCEPTION_GENERATED_TIME_COLUMN_IS_OUT_OF_RANGE_43AB0C2A =
44+
"生成的时间列超出范围。";
45+
public static final String
46+
EXCEPTION_THE_TIME_PREDICATE_DOES_NOT_SELECT_ANY_TIME_RANGE_ARG_98DBCE32 =
47+
"时间谓词未选择任何时间范围：%s";
48+
public static final String
49+
EXCEPTION_RIGHT_CLOSED_GROUP_BY_TIME_WITH_LONG_MAX_VALUE_END_TIME_CANNOT_BE_REPRESENTED_AS_A_SINGLE_RIGHT_OPEN_TIME_FILTER_772F893F =
50+
"右闭 GROUP BY TIME 的结束时间为 Long.MAX_VALUE，无法表示为单个右开时间过滤器。";
51+
public static final String EXCEPTION_DATE_EXPRESSION_IS_OUT_OF_RANGE_ARG_ED35A8A1 =
52+
"日期表达式超出范围：%s";
53+
4154
public static final String TIME_DURATION_WITH_SLIDINGWINDOW_IS_NOT_SUPPORTED_NOW =
4255
"目前不支持 TIME_DURATION 与滑动窗口组合使用";
4356
public static final String MODE_WITH_SLIDINGWINDOW_IS_NOT_SUPPORTED_NOW =

‎iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@ private StorageEngineMessages() {}
4747
public static final String REBOOT_ALL_TIMED_SERVICE_SUCCESSFULLY = "所有定时服务重启成功";
4848
public static final String FAILED_TO_DELETE = "删除失败: {} -> {}";
4949
public static final String FAILED_TO_CHECK_OBJECT_FILES = "检查对象文件失败: {}";
50+
public static final String EXCEPTION_CANNOT_REPAIR_ARG_BECAUSE_TARGET_FILE_ALREADY_EXISTS_AND_TIMESTAMP_IS_LONG_MAX_VALUE_F29F630A =
51+
"无法修复 %s，因为目标文件已存在且文件时间戳为 Long.MAX_VALUE";
5052

5153
// ======================== Buffer Cache ========================
5254

0 commit comments

Comments
 (0)