Skip to content

Commit 61792b4

Browse files
CaideyipiCaideyipi
committed
Fix more library UDF edge cases
1 parent 00f0976 commit 61792b4

33 files changed

Lines changed: 928 additions & 181 deletions

‎library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFLOF.java‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ public class UDTFLOF implements UDTF {
3939
private int multipleK;
4040
private int dim;
4141
private static final String DEFAULT_METHOD = "default";
42+
private static final String METHOD_SERIES = "series";
4243
private String method = DEFAULT_METHOD;
4344
private int window;
4445

@@ -126,6 +127,23 @@ public void validate(UDFParameterValidator validator) throws Exception {
126127
for (int i = 0; i < validator.getParameters().getChildExpressionsSize(); i++) {
127128
validator.validateInputSeriesDataType(i, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE);
128129
}
130+
validator
131+
.validate(
132+
k -> (int) k > 0,
133+
"Parameter k should be a positive integer.",
134+
validator.getParameters().getIntOrDefault("k", 3))
135+
.validate(
136+
window -> (int) window > 0,
137+
"Parameter window should be a positive integer.",
138+
validator.getParameters().getIntOrDefault("window", 10000))
139+
.validate(
140+
method -> isValidMethod((String) method),
141+
"Method should be default or series.",
142+
validator.getParameters().getStringOrDefault("method", DEFAULT_METHOD));
143+
}
144+
145+
private static boolean isValidMethod(String method) {
146+
return DEFAULT_METHOD.equalsIgnoreCase(method) || METHOD_SERIES.equalsIgnoreCase(method);
129147
}
130148

131149
@Override
@@ -143,7 +161,7 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
143161

144162
@Override
145163
public void transform(RowWindow rowWindow, PointCollector collector) throws Exception {
146-
if (this.method.equals(DEFAULT_METHOD)) {
164+
if (this.method.equalsIgnoreCase(DEFAULT_METHOD)) {
147165
int size = 0;
148166
Double[][] knn = new Double[rowWindow.windowSize()][dim];
149167
long[] timestamp = new long[rowWindow.windowSize()];
@@ -181,7 +199,7 @@ public void transform(RowWindow rowWindow, PointCollector collector) throws Exce
181199
}
182200
}
183201
}
184-
} else if (this.method.equals("series")) {
202+
} else if (this.method.equalsIgnoreCase(METHOD_SERIES)) {
185203
int size = rowWindow.windowSize() - window + 1;
186204
if (size > 0) {
187205
List<Long> timestamp = new ArrayList<>();

‎library-udf/src/main/java/org/apache/iotdb/library/dlearn/UDTFAR.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,9 @@ public void terminate(PointCollector collector) throws Exception {
9696
}
9797
}
9898
long interval = maxFreqInterval;
99+
if (interval <= 0) {
100+
return;
101+
}
99102

100103
List<Long> imputedTimeWindow = new ArrayList<>();
101104
List<Double> imputedValueWindow = new ArrayList<>();
@@ -137,7 +140,13 @@ public void terminate(PointCollector collector) throws Exception {
137140
for (int j = 1; j <= i - 1; j++) {
138141
tmpSum += alphas[j][i - 1] * resultCovariances[i - j];
139142
}
143+
if (!Double.isFinite(epsilons[i - 1]) || Math.abs(epsilons[i - 1]) < 1e-12) {
144+
return;
145+
}
140146
kappas[i] = (resultCovariances[i] - tmpSum) / epsilons[i - 1];
147+
if (!Double.isFinite(kappas[i])) {
148+
return;
149+
}
141150
alphas[i][i] = kappas[i];
142151
if (i > 1) {
143152
for (int j = 1; j <= i - 1; j++) {
@@ -147,7 +156,9 @@ public void terminate(PointCollector collector) throws Exception {
147156
epsilons[i] = (1 - kappas[i] * kappas[i]) * epsilons[i - 1];
148157
}
149158
for (int i = 1; i <= p; i++) {
150-
collector.putDouble(i, alphas[i][p]);
159+
if (Double.isFinite(alphas[i][p])) {
160+
collector.putDouble(i, alphas[i][p]);
161+
}
151162
}
152163
}
153164
}

‎library-udf/src/main/java/org/apache/iotdb/library/dmatch/UDAFCov.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -72,11 +72,11 @@ public void transform(Row row, PointCollector collector) throws Exception {
7272

7373
@Override
7474
public void terminate(PointCollector collector) throws Exception {
75-
if (count > 0) { // calculate Cov only when there is more than 1 point
75+
if (count > 0) {
7676
double cov = (sumXY - sumX * sumY / count) / count;
77-
collector.putDouble(0, cov);
78-
} else {
79-
collector.putDouble(0, Double.NaN);
77+
if (Double.isFinite(cov)) {
78+
collector.putDouble(0, cov);
79+
}
8080
}
8181
}
8282
}

‎library-udf/src/main/java/org/apache/iotdb/library/dmatch/UDAFPearson.java‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -78,14 +78,20 @@ public void transform(Row row, PointCollector collector) throws Exception {
7878

7979
@Override
8080
public void terminate(PointCollector collector) throws Exception {
81-
if (count > 0) { // calculate R only when there is more than 1 point
82-
double pearson =
83-
(count * sumXY - sumX * sumY)
84-
/ Math.sqrt(count * sumXX - sumX * sumX)
85-
/ Math.sqrt(count * sumYY - sumY * sumY);
81+
if (count == 0) {
82+
return;
83+
}
84+
double xVariance = count * sumXX - sumX * sumX;
85+
double yVariance = count * sumYY - sumY * sumY;
86+
if (!Double.isFinite(xVariance)
87+
|| !Double.isFinite(yVariance)
88+
|| xVariance <= 0
89+
|| yVariance <= 0) {
90+
return;
91+
}
92+
double pearson = (count * sumXY - sumX * sumY) / Math.sqrt(xVariance) / Math.sqrt(yVariance);
93+
if (Double.isFinite(pearson)) {
8694
collector.putDouble(0, pearson);
87-
} else {
88-
collector.putDouble(0, Double.NaN);
8995
}
9096
}
9197
}

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDAFIntegral.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,9 @@ public void validate(UDFParameterValidator validator) throws Exception {
5050
.validateInputSeriesNumber(1)
5151
.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
5252
.validate(
53-
x -> (long) x > 0,
53+
x -> Util.isPositiveTime((String) x, validator.getParameters()),
5454
"Unknown time unit input. Supported units are ns, us, ms, s, m, h, d.",
55-
Util.parseTime(
56-
validator.getParameters().getStringOrDefault(TIME_UNIT_KEY, TIME_UNIT_S),
57-
validator.getParameters()));
55+
validator.getParameters().getStringOrDefault(TIME_UNIT_KEY, TIME_UNIT_S));
5856
}
5957

6058
@Override

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDAFSkew.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -72,9 +72,16 @@ public void terminate(PointCollector collector) throws Exception {
7272
if (count == 0) {
7373
return;
7474
}
75-
collector.putDouble(
76-
0,
77-
(sumX3 / count - 3 * sumX1 / count * sumX2 / count + 2 * Math.pow(sumX1 / count, 3))
78-
/ Math.pow(sumX2 / count - sumX1 / count * sumX1 / count, 1.5));
75+
double mean = sumX1 / count;
76+
double variance = sumX2 / count - mean * mean;
77+
if (!Double.isFinite(variance) || variance <= 0) {
78+
return;
79+
}
80+
double skew =
81+
(sumX3 / count - 3 * mean * sumX2 / count + 2 * Math.pow(mean, 3))
82+
/ Math.pow(variance, 1.5);
83+
if (Double.isFinite(skew)) {
84+
collector.putDouble(0, skew);
85+
}
7986
}
8087
}

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDTFPACF.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,8 +84,12 @@ public void terminate(PointCollector collector) throws Exception {
8484
x[i] -= xmean;
8585
}
8686
collector.putDouble(timestamp.get(0), 1.0d);
87+
YuleWalker yuleWalker = new YuleWalker();
8788
for (int k = 1; k <= lag; k++) {
88-
collector.putDouble(timestamp.get(k), new YuleWalker().yuleWalker(x, k, method, n));
89+
double partialCorrelation = yuleWalker.yuleWalker(x, k, method, n);
90+
if (Double.isFinite(partialCorrelation)) {
91+
collector.putDouble(timestamp.get(k), partialCorrelation);
92+
}
8993
}
9094
}
9195
}

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDTFResample.java‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,6 @@
3030
import org.apache.iotdb.udf.api.customizer.strategy.RowByRowAccessStrategy;
3131
import org.apache.iotdb.udf.api.type.Type;
3232

33-
import java.text.SimpleDateFormat;
34-
3533
/** This function does upsample or downsample of input series. */
3634
public class UDTFResample implements UDTF {
3735
private static final String START_PARAM = "start";
@@ -45,9 +43,9 @@ public void validate(UDFParameterValidator validator) throws Exception {
4543
.validateInputSeriesDataType(0, Type.DOUBLE, Type.FLOAT, Type.INT32, Type.INT64)
4644
.validateRequiredAttribute("every")
4745
.validate(
48-
x -> (long) x > 0,
46+
x -> Util.isPositiveTime((String) x, validator.getParameters()),
4947
"gap should be a time period whose unit is ms, s, m, h, d.",
50-
Util.parseTime(validator.getParameters().getString("every"), validator.getParameters()))
48+
validator.getParameters().getString("every"))
5149
.validate(
5250
x ->
5351
"min".equals(x)
@@ -62,18 +60,17 @@ public void validate(UDFParameterValidator validator) throws Exception {
6260
x -> "nan".equals(x) || "ffill".equals(x) || "bfill".equals(x) || "linear".equals(x),
6361
"aggr should be min, max, mean, median, first, last.",
6462
validator.getParameters().getStringOrDefault("interp", "nan").toLowerCase());
65-
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
6663
if (validator.getParameters().hasAttribute(START_PARAM)) {
6764
validator.validate(
68-
x -> (long) x > 0,
65+
x -> Util.isPositiveDateTime((String) x),
6966
"start should conform to the format yyyy-MM-dd HH:mm:ss.",
70-
format.parse(validator.getParameters().getString(START_PARAM)).getTime());
67+
validator.getParameters().getString(START_PARAM));
7168
}
7269
if (validator.getParameters().hasAttribute("end")) {
7370
validator.validate(
74-
x -> (long) x > 0,
71+
x -> Util.isPositiveDateTime((String) x),
7572
"end should conform to the format yyyy-MM-dd HH:mm:ss.",
76-
format.parse(validator.getParameters().getString("end")).getTime());
73+
validator.getParameters().getString("end"));
7774
}
7875
}
7976

@@ -84,14 +81,13 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
8481
long newPeriod = Util.parseTime(parameters.getString("every"), parameters);
8582
String aggregator = parameters.getStringOrDefault("aggr", "mean").toLowerCase();
8683
String interpolator = parameters.getStringOrDefault("interp", "nan").toLowerCase();
87-
SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
8884
long startTime = -1;
8985
long endTime = -1;
9086
if (parameters.hasAttribute(START_PARAM)) {
91-
startTime = format.parse(parameters.getString(START_PARAM)).getTime();
87+
startTime = Util.parseDateTime(parameters.getString(START_PARAM));
9288
}
9389
if (parameters.hasAttribute("end")) {
94-
endTime = format.parse(parameters.getString("end")).getTime();
90+
endTime = Util.parseDateTime(parameters.getString("end"));
9591
}
9692
resampler = new Resampler(newPeriod, aggregator, interpolator, startTime, endTime);
9793
}

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDTFSegment.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,9 @@ public void transform(Row row, PointCollector collector) throws Exception {
9898

9999
@Override
100100
public void terminate(PointCollector collector) throws Exception {
101+
if (value.isEmpty()) {
102+
return;
103+
}
101104
long[] ts = timestamp.stream().mapToLong(Long::valueOf).toArray();
102105
double[] v = value.stream().mapToDouble(Double::valueOf).toArray();
103106
List<double[]> seg = new ArrayList<>();

‎library-udf/src/main/java/org/apache/iotdb/library/dprofile/UDTFSpline.java‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -78,9 +78,11 @@ public void transform(Row row, PointCollector collector) throws Exception {
7878
Long t = row.getTime();
7979
if (minimumTimestamp == null) {
8080
minimumTimestamp = t;
81+
} else if (t <= timestamp.get(timestamp.size() - 1)) {
82+
return;
8183
}
8284
timestamp.add(t);
83-
xDouble.add((Double.valueOf(Long.toString(t - minimumTimestamp))));
85+
xDouble.add((double) t - minimumTimestamp);
8486
yDouble.add(v);
8587
}
8688
}
@@ -93,14 +95,12 @@ public void terminate(PointCollector collector) throws Exception {
9395
double[] y = ArrayUtils.toPrimitive(yDouble.toArray(new Double[0]));
9496
psf = asi.interpolate(x, y);
9597
for (int i = 0; i < samplePoints; i++) {
96-
int approximation =
97-
(int)
98-
Math.floor(
99-
(x[0] * (samplePoints - 1 - i) + x[yDouble.size() - 1] * (i))
100-
/ (samplePoints - 1)
101-
+ 0.5);
98+
double approximation =
99+
Math.floor(
100+
(x[0] * (samplePoints - 1 - i) + x[yDouble.size() - 1] * (i)) / (samplePoints - 1)
101+
+ 0.5);
102102
double yhead = psf.value(approximation);
103-
collector.putDouble(minimumTimestamp + approximation, yhead);
103+
collector.putDouble(minimumTimestamp + Math.round(approximation), yhead);
104104
}
105105
}
106106
}

0 commit comments

Comments
 (0)