Skip to content

Commit 36af775

Browse files
lbooker42lbooker42
authored
Add exception for exponential operators providing non-descending timestamps. (deephaven#5144)
* Add exception for exponential operators non-descending timestamps. * Changed exception message. * Changed exception messages and improved comments.
1 parent fc537c9 commit 36af775

20 files changed

+221
-134
lines changed

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/UpdateByBucketHelper.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -282,7 +282,7 @@ private int fillChunkWithNonNull(LongChunk<OrderedRowKeys> keysChunk, LongChunk<
282282
}
283283
if (ts < lastTimestamp.longValue()) {
284284
throw (new TableDataException(
285-
"updateBy time-based operators require non-descending timestamp values"));
285+
"Timestamp values in UpdateBy operators must not decrease"));
286286
}
287287

288288
ssaValues.add(ts);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/BasePrimitiveEMOperator.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,8 @@ public void initializeCumulative(
9494
ctx.lastStamp = firstUnmodifiedTimestamp;
9595
}
9696

97-
void handleBadData(@NotNull final Context ctx,
97+
void handleBadData(
98+
@NotNull final Context ctx,
9899
final boolean isNull,
99100
final boolean isNan) {
100101
boolean doReset = false;

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/BigDecimalEMOperator.java

+23-18
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import io.deephaven.chunk.attributes.Values;
77
import io.deephaven.engine.rowset.RowSequence;
88
import io.deephaven.engine.table.impl.MatchPair;
9+
import io.deephaven.engine.table.impl.locations.TableDataException;
910
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1011
import org.jetbrains.annotations.NotNull;
1112
import org.jetbrains.annotations.Nullable;
@@ -21,10 +22,11 @@ protected Context(final int chunkSize) {
2122
}
2223

2324
@Override
24-
public void accumulateCumulative(@NotNull RowSequence inputKeys,
25-
Chunk<? extends Values>[] valueChunkArr,
26-
LongChunk<? extends Values> tsChunk,
27-
int len) {
25+
public void accumulateCumulative(
26+
@NotNull final RowSequence inputKeys,
27+
@NotNull final Chunk<? extends Values>[] valueChunkArr,
28+
final LongChunk<? extends Values> tsChunk,
29+
final int len) {
2830
setValueChunks(valueChunkArr);
2931

3032
// chunk processing
@@ -56,22 +58,25 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
5658
handleBadData(this, true);
5759
} else if (isNullTime) {
5860
// no change to curVal and lastStamp
61+
} else if (curVal == null) {
62+
// We have a valid input value, we can initialize the output value with it.
63+
curVal = input;
64+
lastStamp = timestamp;
5965
} else {
60-
if (curVal == null) {
61-
curVal = input;
62-
lastStamp = timestamp;
63-
} else {
64-
final long dt = timestamp - lastStamp;
65-
if (dt != 0) {
66-
// alpha is dynamic based on time, but only recalculated when needed
67-
if (dt != lastDt) {
68-
alpha = computeAlpha(-dt, reverseWindowScaleUnits);
69-
oneMinusAlpha = computeOneMinusAlpha(alpha);
70-
lastDt = dt;
71-
}
72-
curVal = aggFunction.apply(curVal, input, alpha, oneMinusAlpha);
73-
lastStamp = timestamp;
66+
final long dt = timestamp - lastStamp;
67+
if (dt < 0) {
68+
// negative time deltas are not allowed, throw an exception
69+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
70+
}
71+
if (dt != 0) {
72+
// alpha is dynamic based on time, but only recalculated when needed
73+
if (dt != lastDt) {
74+
alpha = computeAlpha(-dt, reverseWindowScaleUnits);
75+
oneMinusAlpha = computeOneMinusAlpha(alpha);
76+
lastDt = dt;
7477
}
78+
curVal = aggFunction.apply(curVal, input, alpha, oneMinusAlpha);
79+
lastStamp = timestamp;
7580
}
7681
}
7782
outputValues.set(ii, curVal);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/BigIntegerEMOperator.java

+23-18
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
import io.deephaven.chunk.attributes.Values;
77
import io.deephaven.engine.rowset.RowSequence;
88
import io.deephaven.engine.table.impl.MatchPair;
9+
import io.deephaven.engine.table.impl.locations.TableDataException;
910
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1011
import org.jetbrains.annotations.NotNull;
1112
import org.jetbrains.annotations.Nullable;
@@ -22,10 +23,11 @@ protected Context(final int chunkSize) {
2223
}
2324

2425
@Override
25-
public void accumulateCumulative(@NotNull RowSequence inputKeys,
26-
Chunk<? extends Values>[] valueChunkArr,
27-
LongChunk<? extends Values> tsChunk,
28-
int len) {
26+
public void accumulateCumulative(
27+
@NotNull final RowSequence inputKeys,
28+
@NotNull final Chunk<? extends Values>[] valueChunkArr,
29+
@Nullable final LongChunk<? extends Values> tsChunk,
30+
final int len) {
2931
setValueChunks(valueChunkArr);
3032

3133
// chunk processing
@@ -58,22 +60,25 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
5860
handleBadData(this, true);
5961
} else if (isNullTime) {
6062
// no change to curVal and lastStamp
63+
} else if (curVal == null) {
64+
// We have a valid input value, we can initialize the output value with it.
65+
curVal = new BigDecimal(input, control.bigValueContextOrDefault());
66+
lastStamp = timestamp;
6167
} else {
62-
final BigDecimal decimalInput = new BigDecimal(input, control.bigValueContextOrDefault());
63-
if (curVal == null) {
64-
curVal = decimalInput;
65-
lastStamp = timestamp;
66-
} else {
67-
final long dt = timestamp - lastStamp;
68-
// Alpha is dynamic based on time, but only recalculated when needed
69-
if (dt != lastDt) {
70-
alpha = computeAlpha(-dt, reverseWindowScaleUnits);
71-
oneMinusAlpha = computeOneMinusAlpha(alpha);
72-
lastDt = dt;
73-
}
74-
curVal = aggFunction.apply(curVal, decimalInput, alpha, oneMinusAlpha);
75-
lastStamp = timestamp;
68+
final long dt = timestamp - lastStamp;
69+
if (dt < 0) {
70+
// negative time deltas are not allowed, throw an exception
71+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
72+
}
73+
// Alpha is dynamic based on time, but only recalculated when needed
74+
if (dt != lastDt) {
75+
alpha = computeAlpha(-dt, reverseWindowScaleUnits);
76+
oneMinusAlpha = computeOneMinusAlpha(alpha);
77+
lastDt = dt;
7678
}
79+
final BigDecimal decimalInput = new BigDecimal(input, control.bigValueContextOrDefault());
80+
curVal = aggFunction.apply(curVal, decimalInput, alpha, oneMinusAlpha);
81+
lastStamp = timestamp;
7782
}
7883
outputValues.set(ii, curVal);
7984
}

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/ByteEMOperator.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import io.deephaven.chunk.attributes.Values;
1313
import io.deephaven.engine.rowset.RowSequence;
1414
import io.deephaven.engine.table.impl.MatchPair;
15+
import io.deephaven.engine.table.impl.locations.TableDataException;
1516
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1617
import org.jetbrains.annotations.NotNull;
1718
import org.jetbrains.annotations.Nullable;
@@ -75,10 +76,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
7576
} else if (isNullTime) {
7677
// no change to curVal and lastStamp
7778
} else if (curVal == NULL_DOUBLE) {
79+
// We have a valid input value, we can initialize the output value with it.
7880
curVal = input;
7981
lastStamp = timestamp;
8082
} else {
8183
final long dt = timestamp - lastStamp;
84+
if (dt < 0) {
85+
// negative time deltas are not allowed, throw an exception
86+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
87+
}
8288
if (dt != lastDt) {
8389
// Alpha is dynamic based on time, but only recalculated when needed
8490
alpha = Math.exp(-dt / reverseWindowScaleUnits);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/CharEMOperator.java

+6
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
import io.deephaven.chunk.attributes.Values;
88
import io.deephaven.engine.rowset.RowSequence;
99
import io.deephaven.engine.table.impl.MatchPair;
10+
import io.deephaven.engine.table.impl.locations.TableDataException;
1011
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1112
import org.jetbrains.annotations.NotNull;
1213
import org.jetbrains.annotations.Nullable;
@@ -69,10 +70,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
6970
} else if (isNullTime) {
7071
// no change to curVal and lastStamp
7172
} else if (curVal == NULL_DOUBLE) {
73+
// We have a valid input value, we can initialize the output value with it.
7274
curVal = input;
7375
lastStamp = timestamp;
7476
} else {
7577
final long dt = timestamp - lastStamp;
78+
if (dt < 0) {
79+
// negative time deltas are not allowed, throw an exception
80+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
81+
}
7682
if (dt != lastDt) {
7783
// Alpha is dynamic based on time, but only recalculated when needed
7884
alpha = Math.exp(-dt / reverseWindowScaleUnits);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/DoubleEMOperator.java

+11-5
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import io.deephaven.chunk.attributes.Values;
1414
import io.deephaven.engine.rowset.RowSequence;
1515
import io.deephaven.engine.table.impl.MatchPair;
16+
import io.deephaven.engine.table.impl.locations.TableDataException;
1617
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1718
import org.jetbrains.annotations.NotNull;
1819
import org.jetbrains.annotations.Nullable;
@@ -28,10 +29,11 @@ protected Context(final int chunkSize) {
2829
}
2930

3031
@Override
31-
public void accumulateCumulative(@NotNull RowSequence inputKeys,
32-
Chunk<? extends Values>[] valueChunkArr,
33-
LongChunk<? extends Values> tsChunk,
34-
int len) {
32+
public void accumulateCumulative(
33+
@NotNull final RowSequence inputKeys,
34+
@NotNull final Chunk<? extends Values>[] valueChunkArr,
35+
final LongChunk<? extends Values> tsChunk,
36+
final int len) {
3537
setValueChunks(valueChunkArr);
3638

3739
// chunk processing
@@ -69,11 +71,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
6971
} else if (isNullTime) {
7072
// no change to curVal and lastStamp
7173
} else if (curVal == NULL_DOUBLE) {
72-
// If the data looks good, and we have a null ema, just accept the current value
74+
// If the data looks good, and we have a null computed value, accept the current value
7375
curVal = input;
7476
lastStamp = timestamp;
7577
} else {
7678
final long dt = timestamp - lastStamp;
79+
if (dt < 0) {
80+
// negative time deltas are not allowed, throw an exception
81+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
82+
}
7783
if (dt != 0) {
7884
final double alpha = Math.exp(-dt / reverseWindowScaleUnits);
7985
final double oneMinusAlpha = 1.0 - alpha;

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/FloatEMOperator.java

+11-5
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import io.deephaven.chunk.attributes.Values;
99
import io.deephaven.engine.rowset.RowSequence;
1010
import io.deephaven.engine.table.impl.MatchPair;
11+
import io.deephaven.engine.table.impl.locations.TableDataException;
1112
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1213
import org.jetbrains.annotations.NotNull;
1314
import org.jetbrains.annotations.Nullable;
@@ -23,10 +24,11 @@ protected Context(final int chunkSize) {
2324
}
2425

2526
@Override
26-
public void accumulateCumulative(@NotNull RowSequence inputKeys,
27-
Chunk<? extends Values>[] valueChunkArr,
28-
LongChunk<? extends Values> tsChunk,
29-
int len) {
27+
public void accumulateCumulative(
28+
@NotNull final RowSequence inputKeys,
29+
@NotNull final Chunk<? extends Values>[] valueChunkArr,
30+
final LongChunk<? extends Values> tsChunk,
31+
final int len) {
3032
setValueChunks(valueChunkArr);
3133

3234
// chunk processing
@@ -64,11 +66,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
6466
} else if (isNullTime) {
6567
// no change to curVal and lastStamp
6668
} else if (curVal == NULL_DOUBLE) {
67-
// If the data looks good, and we have a null ema, just accept the current value
69+
// If the data looks good, and we have a null computed value, accept the current value
6870
curVal = input;
6971
lastStamp = timestamp;
7072
} else {
7173
final long dt = timestamp - lastStamp;
74+
if (dt < 0) {
75+
// negative time deltas are not allowed, throw an exception
76+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
77+
}
7278
if (dt != 0) {
7379
final double alpha = Math.exp(-dt / reverseWindowScaleUnits);
7480
final double oneMinusAlpha = 1.0 - alpha;

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/IntEMOperator.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import io.deephaven.chunk.attributes.Values;
1313
import io.deephaven.engine.rowset.RowSequence;
1414
import io.deephaven.engine.table.impl.MatchPair;
15+
import io.deephaven.engine.table.impl.locations.TableDataException;
1516
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1617
import org.jetbrains.annotations.NotNull;
1718
import org.jetbrains.annotations.Nullable;
@@ -74,10 +75,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
7475
} else if (isNullTime) {
7576
// no change to curVal and lastStamp
7677
} else if (curVal == NULL_DOUBLE) {
78+
// We have a valid input value, we can initialize the output value with it.
7779
curVal = input;
7880
lastStamp = timestamp;
7981
} else {
8082
final long dt = timestamp - lastStamp;
83+
if (dt < 0) {
84+
// negative time deltas are not allowed, throw an exception
85+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
86+
}
8187
if (dt != lastDt) {
8288
// Alpha is dynamic based on time, but only recalculated when needed
8389
alpha = Math.exp(-dt / reverseWindowScaleUnits);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/LongEMOperator.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import io.deephaven.chunk.attributes.Values;
1313
import io.deephaven.engine.rowset.RowSequence;
1414
import io.deephaven.engine.table.impl.MatchPair;
15+
import io.deephaven.engine.table.impl.locations.TableDataException;
1516
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1617
import org.jetbrains.annotations.NotNull;
1718
import org.jetbrains.annotations.Nullable;
@@ -74,10 +75,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
7475
} else if (isNullTime) {
7576
// no change to curVal and lastStamp
7677
} else if (curVal == NULL_DOUBLE) {
78+
// We have a valid input value, we can initialize the output value with it.
7779
curVal = input;
7880
lastStamp = timestamp;
7981
} else {
8082
final long dt = timestamp - lastStamp;
83+
if (dt < 0) {
84+
// negative time deltas are not allowed, throw an exception
85+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
86+
}
8187
if (dt != lastDt) {
8288
// Alpha is dynamic based on time, but only recalculated when needed
8389
alpha = Math.exp(-dt / reverseWindowScaleUnits);

‎engine/table/src/main/java/io/deephaven/engine/table/impl/updateby/em/ShortEMOperator.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import io.deephaven.chunk.attributes.Values;
1313
import io.deephaven.engine.rowset.RowSequence;
1414
import io.deephaven.engine.table.impl.MatchPair;
15+
import io.deephaven.engine.table.impl.locations.TableDataException;
1516
import io.deephaven.engine.table.impl.updateby.UpdateByOperator;
1617
import org.jetbrains.annotations.NotNull;
1718
import org.jetbrains.annotations.Nullable;
@@ -74,10 +75,15 @@ public void accumulateCumulative(@NotNull RowSequence inputKeys,
7475
} else if (isNullTime) {
7576
// no change to curVal and lastStamp
7677
} else if (curVal == NULL_DOUBLE) {
78+
// We have a valid input value, we can initialize the output value with it.
7779
curVal = input;
7880
lastStamp = timestamp;
7981
} else {
8082
final long dt = timestamp - lastStamp;
83+
if (dt < 0) {
84+
// negative time deltas are not allowed, throw an exception
85+
throw new TableDataException("Timestamp values in UpdateBy operators must not decrease");
86+
}
8187
if (dt != lastDt) {
8288
// Alpha is dynamic based on time, but only recalculated when needed
8389
alpha = Math.exp(-dt / reverseWindowScaleUnits);

0 commit comments

Comments
 (0)