Skip to content

Commit 7b1b673

Browse files
CaideyipiCaideyipi
committed
Fix library UDF edge cases
1 parent eb4409d commit 7b1b673

48 files changed

Lines changed: 1416 additions & 162 deletions

Some content is hidden

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

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

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,12 @@ public void validate(UDFParameterValidator validator) throws Exception {
6363
"parameter $q1$ should be smaller than $q3$",
6464
validator.getParameters().getDoubleOrDefault("q1", -1),
6565
validator.getParameters().getDoubleOrDefault("q3", 1));
66+
if (validator
67+
.getParameters()
68+
.getStringOrDefault("compute", BATCH_COMPUTE)
69+
.equalsIgnoreCase(STREAM_COMPUTE)) {
70+
validator.validateRequiredAttribute("q1").validateRequiredAttribute("q3");
71+
}
6672
}
6773

6874
@Override
@@ -91,14 +97,19 @@ public void transform(Row row, PointCollector collector) throws Exception {
9197
}
9298
} else if (compute.equalsIgnoreCase(BATCH_COMPUTE)) {
9399
double v = Util.getValueAsDouble(row);
94-
value.add(v);
95-
timestamp.add(row.getTime());
100+
if (Double.isFinite(v)) {
101+
value.add(v);
102+
timestamp.add(row.getTime());
103+
}
96104
}
97105
}
98106

99107
@Override
100108
public void terminate(PointCollector collector) throws Exception {
101109
if (compute.equalsIgnoreCase(BATCH_COMPUTE)) {
110+
if (value.isEmpty()) {
111+
return;
112+
}
102113
q1 = Quantiles.quartiles().index(1).compute(value);
103114
q3 = Quantiles.quartiles().index(3).compute(value);
104115
iqr = q3 - q1;

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

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@
3333

3434
/** This function detects outliers which lies over average +/- k * sigma. */
3535
public class UDTFKSigma implements UDTF {
36+
private static final int DEFAULT_WINDOW_SIZE = 10;
37+
3638
private double mean = 0.0;
3739
private double variance = 0.0;
3840
private double sumX2 = 0.0;
@@ -51,7 +53,7 @@ public void validate(UDFParameterValidator validator) throws Exception {
5153
.validate(
5254
x -> (int) x > 0,
5355
"Window size should be larger than 0.",
54-
validator.getParameters().getIntOrDefault("window", 10))
56+
validator.getParameters().getIntOrDefault("window", DEFAULT_WINDOW_SIZE))
5557
.validate(
5658
x -> (double) x > 0,
5759
"Parameter k should be larger than 0.",
@@ -66,7 +68,7 @@ public void beforeStart(UDFParameters udfParameters, UDTFConfigurations udtfConf
6668
.setOutputDataType(udfParameters.getDataType(0));
6769
this.multipleK = udfParameters.getDoubleOrDefault("k", 3);
6870
this.dataType = udfParameters.getDataType(0);
69-
this.windowSize = udfParameters.getIntOrDefault("window", 10000);
71+
this.windowSize = udfParameters.getIntOrDefault("window", DEFAULT_WINDOW_SIZE);
7072
this.v = new CircularQueue<>(windowSize);
7173
this.t = new LongCircularQueue(windowSize);
7274
}

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

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,10 @@ private double dist(Double[] nnk, Double[] x) {
119119

120120
@Override
121121
public void validate(UDFParameterValidator validator) throws Exception {
122-
validator.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE);
122+
validator.validateInputSeriesNumber(1, Integer.MAX_VALUE);
123+
for (int i = 0; i < validator.getParameters().getChildExpressionsSize(); i++) {
124+
validator.validateInputSeriesDataType(i, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE);
125+
}
123126
}
124127

125128
@Override
@@ -147,7 +150,7 @@ public void transform(RowWindow rowWindow, PointCollector collector) throws Exce
147150
timestamp[i] = rowWindow.getRow(row).getTime();
148151
for (int j = 0; j < dim; j++) {
149152
if (!rowWindow.getRow(row).isNull(j)) {
150-
knn[i][j] = Util.getValueAsDouble(rowWindow.getRow(i), j);
153+
knn[i][j] = Util.getValueAsDouble(rowWindow.getRow(row), j);
151154
} else {
152155
i--;
153156
size--;

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

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,14 @@ public class UDTFRange implements UDTF {
4040
public void validate(UDFParameterValidator validator) throws Exception {
4141
validator
4242
.validateInputSeriesNumber(1)
43-
.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE);
43+
.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
44+
.validateRequiredAttribute("lower_bound")
45+
.validateRequiredAttribute("upper_bound")
46+
.validate(
47+
params -> (double) params[0] < (double) params[1],
48+
"parameter \"lower_bound\" should be smaller than \"upper_bound\".",
49+
validator.getParameters().getDouble("lower_bound"),
50+
validator.getParameters().getDouble("upper_bound"));
4451
}
4552

4653
@Override

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.iotdb.udf.api.access.Row;
2626
import org.apache.iotdb.udf.api.collector.PointCollector;
2727
import org.apache.iotdb.udf.api.customizer.config.UDTFConfigurations;
28+
import org.apache.iotdb.udf.api.customizer.parameter.UDFParameterValidator;
2829
import org.apache.iotdb.udf.api.customizer.parameter.UDFParameters;
2930
import org.apache.iotdb.udf.api.customizer.strategy.RowByRowAccessStrategy;
3031
import org.apache.iotdb.udf.api.exception.UDFException;
@@ -38,12 +39,25 @@ public class UDTFAR implements UDTF {
3839
private List<Long> timeWindow = new ArrayList<>();
3940
private List<Double> valueWindow = new ArrayList<>();
4041

42+
@Override
43+
public void validate(UDFParameterValidator validator) throws Exception {
44+
validator
45+
.validateInputSeriesNumber(1)
46+
.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
47+
.validate(
48+
x -> (int) x > 0,
49+
"Parameter p should be a positive integer.",
50+
validator.getParameters().getIntOrDefault("p", 1));
51+
}
52+
4153
@Override
4254
public void beforeStart(UDFParameters udfParameters, UDTFConfigurations udtfConfigurations)
4355
throws Exception {
4456
udtfConfigurations
4557
.setAccessStrategy(new RowByRowAccessStrategy())
4658
.setOutputDataType(udfParameters.getDataType(0));
59+
timeWindow.clear();
60+
valueWindow.clear();
4761
this.p = udfParameters.getIntOrDefault("p", 1);
4862
udtfConfigurations.setAccessStrategy(new RowByRowAccessStrategy());
4963
udtfConfigurations.setOutputDataType(Type.DOUBLE);

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,8 @@ public void validate(UDFParameterValidator validator) throws Exception {
7272
validator
7373
.validateInputSeriesNumber(1)
7474
.validateInputSeriesDataType(0, Type.INT32, Type.INT64, Type.FLOAT, Type.DOUBLE)
75+
.validateRequiredAttribute("l")
76+
.validateRequiredAttribute("k")
7577
.validate(
7678
x -> (int) x > 0,
7779
"Parameter l must be a positive integer.",

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

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ public void beforeStart(UDFParameters parameters, UDTFConfigurations configurati
5252
configurations
5353
.setAccessStrategy(new SlidingSizeWindowAccessStrategy(Integer.MAX_VALUE))
5454
.setOutputDataType(Type.DOUBLE);
55+
dp = null;
56+
m = 0;
5557
}
5658

5759
@Override
@@ -68,6 +70,10 @@ public void transform(RowWindow rowWindow, PointCollector collector) throws Exce
6870
b.add(Util.getValueAsDouble(row, 1));
6971
}
7072
m = a.size();
73+
if (m == 0) {
74+
dp = null;
75+
return;
76+
}
7177
dp = new double[m + 1][m + 1];
7278
for (int i = 1; i <= m; i++) {
7379
dp[0][i] = dp[i][0] = Double.MAX_VALUE;
@@ -84,6 +90,8 @@ public void transform(RowWindow rowWindow, PointCollector collector) throws Exce
8490

8591
@Override
8692
public void terminate(PointCollector collector) throws Exception {
87-
collector.putDouble(0, dp[m][m]);
93+
if (dp != null) {
94+
collector.putDouble(0, dp[m][m]);
95+
}
8896
}
8997
}

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

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

34+
import java.util.NoSuchElementException;
35+
3436
/** calculate the exact or approximate median absolute deviation (mad). */
3537
public class UDAFMad implements UDTF {
3638

@@ -73,10 +75,14 @@ public void transform(Row row, PointCollector collector) throws Exception {
7375

7476
@Override
7577
public void terminate(PointCollector collector) throws Exception {
76-
if (exact) {
77-
collector.putDouble(0, statistics.getMad());
78-
} else {
79-
collector.putDouble(0, sketch.getMad().result);
78+
try {
79+
if (exact) {
80+
collector.putDouble(0, statistics.getMad());
81+
} else {
82+
collector.putDouble(0, sketch.getMad().result);
83+
}
84+
} catch (NoSuchElementException e) {
85+
// Empty inputs have no MAD to emit.
8086
}
8187
}
8288
}

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -81,8 +81,8 @@ public void terminate(PointCollector collector) throws Exception {
8181
} else {
8282
collector.putDouble(0, sketch.query(0.5));
8383
}
84-
} catch (NoSuchElementException e) {
85-
// just ignore it
84+
} catch (NoSuchElementException | ArithmeticException e) {
85+
// Empty inputs have no median to emit.
8686
}
8787
}
8888
}

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

Lines changed: 76 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -33,13 +33,14 @@
3333

3434
import java.util.HashMap;
3535
import java.util.Map;
36+
import java.util.NoSuchElementException;
3637

3738
/** calculate the approximate percentile. */
3839
public class UDAFPercentile implements UDTF {
39-
protected static Map<Integer, Long> intDic;
40-
protected static Map<Long, Long> longDic;
41-
protected static Map<Float, Long> floatDic;
42-
protected static Map<Double, Long> doubleDic;
40+
protected Map<Integer, Long> intDic;
41+
protected Map<Long, Long> longDic;
42+
protected Map<Float, Long> floatDic;
43+
protected Map<Double, Long> doubleDic;
4344
private ExactOrderStatistics statistics;
4445
private GKArray sketch;
4546
private boolean exact;
@@ -106,16 +107,24 @@ public void transform(Row row, PointCollector collector) throws Exception {
106107
statistics.insert(row);
107108
switch (dataType) {
108109
case INT32:
109-
intDic.put(row.getInt(0), row.getTime());
110+
if (!row.isNull(0)) {
111+
intDic.put(row.getInt(0), row.getTime());
112+
}
110113
break;
111114
case INT64:
112-
longDic.put(row.getLong(0), row.getTime());
115+
if (!row.isNull(0)) {
116+
longDic.put(row.getLong(0), row.getTime());
117+
}
113118
break;
114119
case FLOAT:
115-
floatDic.put(row.getFloat(0), row.getTime());
120+
if (!row.isNull(0) && Float.isFinite(row.getFloat(0))) {
121+
floatDic.put(row.getFloat(0), row.getTime());
122+
}
116123
break;
117124
case DOUBLE:
118-
doubleDic.put(row.getDouble(0), row.getTime());
125+
if (!row.isNull(0) && Double.isFinite(row.getDouble(0))) {
126+
doubleDic.put(row.getDouble(0), row.getTime());
127+
}
119128
break;
120129
case BLOB:
121130
case BOOLEAN:
@@ -134,62 +143,66 @@ public void transform(Row row, PointCollector collector) throws Exception {
134143

135144
@Override
136145
public void terminate(PointCollector collector) throws Exception {
137-
if (exact) {
138-
long time;
139-
switch (dataType) {
140-
case INT32:
141-
int ires = Integer.parseInt(statistics.getPercentile(rank));
142-
time = intDic.getOrDefault(ires, 0L);
143-
collector.putInt(time, ires);
144-
break;
145-
case INT64:
146-
long lres = Long.parseLong(statistics.getPercentile(rank));
147-
time = longDic.getOrDefault(lres, 0L);
148-
collector.putLong(time, lres);
149-
break;
150-
case FLOAT:
151-
float fres = Float.parseFloat(statistics.getPercentile(rank));
152-
time = floatDic.getOrDefault(fres, 0L);
153-
collector.putFloat(time, fres);
154-
break;
155-
case DOUBLE:
156-
double dres = Double.parseDouble(statistics.getPercentile(rank));
157-
time = doubleDic.getOrDefault(dres, 0L);
158-
collector.putDouble(time, dres);
159-
break;
160-
case DATE:
161-
case TIMESTAMP:
162-
case TEXT:
163-
case STRING:
164-
case BOOLEAN:
165-
case BLOB:
166-
default:
167-
break;
168-
}
169-
} else {
170-
double res = sketch.query(rank);
171-
switch (dataType) {
172-
case INT32:
173-
collector.putInt(0, (int) res);
174-
break;
175-
case INT64:
176-
collector.putLong(0, (long) res);
177-
break;
178-
case FLOAT:
179-
collector.putFloat(0, (float) res);
180-
break;
181-
case DOUBLE:
182-
collector.putDouble(0, res);
183-
break;
184-
case BOOLEAN:
185-
case BLOB:
186-
case STRING:
187-
case TEXT:
188-
case TIMESTAMP:
189-
case DATE:
190-
default:
191-
break;
146+
try {
147+
if (exact) {
148+
long time;
149+
switch (dataType) {
150+
case INT32:
151+
int ires = Integer.parseInt(statistics.getPercentile(rank));
152+
time = intDic.getOrDefault(ires, 0L);
153+
collector.putInt(time, ires);
154+
break;
155+
case INT64:
156+
long lres = Long.parseLong(statistics.getPercentile(rank));
157+
time = longDic.getOrDefault(lres, 0L);
158+
collector.putLong(time, lres);
159+
break;
160+
case FLOAT:
161+
float fres = Float.parseFloat(statistics.getPercentile(rank));
162+
time = floatDic.getOrDefault(fres, 0L);
163+
collector.putFloat(time, fres);
164+
break;
165+
case DOUBLE:
166+
double dres = Double.parseDouble(statistics.getPercentile(rank));
167+
time = doubleDic.getOrDefault(dres, 0L);
168+
collector.putDouble(time, dres);
169+
break;
170+
case DATE:
171+
case TIMESTAMP:
172+
case TEXT:
173+
case STRING:
174+
case BOOLEAN:
175+
case BLOB:
176+
default:
177+
break;
178+
}
179+
} else {
180+
double res = sketch.query(rank);
181+
switch (dataType) {
182+
case INT32:
183+
collector.putInt(0, (int) res);
184+
break;
185+
case INT64:
186+
collector.putLong(0, (long) res);
187+
break;
188+
case FLOAT:
189+
collector.putFloat(0, (float) res);
190+
break;
191+
case DOUBLE:
192+
collector.putDouble(0, res);
193+
break;
194+
case BOOLEAN:
195+
case BLOB:
196+
case STRING:
197+
case TEXT:
198+
case TIMESTAMP:
199+
case DATE:
200+
default:
201+
break;
202+
}
192203
}
204+
} catch (NoSuchElementException | ArithmeticException e) {
205+
// Empty inputs have no percentile to emit.
193206
}
194207
}
195208
}

0 commit comments

Comments
 (0)