Skip to content

Commit 28919e9

Browse files
CaideyipiCaideyipi
committed
Fix library UDF matching edge cases
1 parent 61792b4 commit 28919e9

5 files changed

Lines changed: 167 additions & 26 deletions

File tree

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -92,9 +92,9 @@ public void transform(RowWindow rowWindow, PointCollector collector) throws Exce
9292
}
9393
}
9494
for (int len = 3; len <= m; len++) {
95-
for (int i = 1, j = len; j <= m; j++) {
95+
for (int i = 1, j = len; j <= m; i++, j++) {
9696
dp[i][j] =
97-
Math.pow(Math.abs(a.get(0) - a.get(j - 1)), 2)
97+
Math.pow(Math.abs(a.get(i - 1) - a.get(j - 1)), 2)
9898
+ Math.min(Math.min(dp[i + 1][j], dp[i][j - 1]), dp[i + 1][j - 1]);
9999
}
100100
}

‎library-udf/src/main/java/org/apache/iotdb/library/match/UDAFDTWMatch.java‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -118,25 +118,27 @@ private double getValue(Column column, int i) {
118118

119119
private float calculateDTW(Double[] series1, Double[] series2) {
120120
int n = series1.length;
121-
double[][] dtw = new double[n][n];
121+
int m = series2.length;
122+
if (n == 0 || m == 0) {
123+
return Float.POSITIVE_INFINITY;
124+
}
122125

123-
// Initialize the DTW matrix
124-
for (int i = 0; i < n; i++) {
125-
for (int j = 0; j < n; j++) {
126+
double[][] dtw = new double[n + 1][m + 1];
127+
for (int i = 0; i <= n; i++) {
128+
for (int j = 0; j <= m; j++) {
126129
dtw[i][j] = Double.POSITIVE_INFINITY;
127130
}
128131
}
129132
dtw[0][0] = 0;
130133

131-
// Compute the DTW distance
132-
for (int i = 1; i < n; i++) {
133-
for (int j = 1; j < n; j++) {
134-
double cost = Math.abs(series1[i] - series2[j]);
134+
for (int i = 1; i <= n; i++) {
135+
for (int j = 1; j <= m; j++) {
136+
double cost = Math.abs(series1[i - 1] - series2[j - 1]);
135137
dtw[i][j] = cost + Math.min(Math.min(dtw[i - 1][j], dtw[i][j - 1]), dtw[i - 1][j - 1]);
136138
}
137139
}
138140

139-
return (float) dtw[n - 1][n - 1];
141+
return (float) dtw[n][m];
140142
}
141143

142144
@Override

‎library-udf/src/main/java/org/apache/iotdb/library/match/UDAFPatternMatch.java‎

Lines changed: 36 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040

4141
import java.nio.charset.Charset;
4242
import java.util.Arrays;
43+
import java.util.Collections;
4344
import java.util.List;
4445
import java.util.Map;
4546
import java.util.stream.IntStream;
@@ -113,17 +114,20 @@ public void outputFinal(State state, ResultValue resultValue) {
113114
resultValue.setNull();
114115
return;
115116
}
116-
PatternExecutor executor = new PatternExecutor();
117117

118-
List<Point> sourcePointsExtract = executor.scalePoint(times, values);
119-
List<Point> queryPointsExtract = executor.extractPoints(timePattern, valuePattern);
120-
121-
executor.setPoints(queryPointsExtract);
122-
PatternContext ctx = new PatternContext();
123-
ctx.setThreshold(threshold);
124-
ctx.setDataPoints(sourcePointsExtract);
125-
// State only records time and recorded values, and the final result is calculated
126-
List<PatternResult> results = executor.executeQuery(ctx);
118+
List<PatternResult> results = Collections.emptyList();
119+
if (hasPositiveTimeRange(times) && hasPositiveValueRange(values)) {
120+
PatternExecutor executor = new PatternExecutor();
121+
List<Point> sourcePointsExtract = executor.scalePoint(times, values);
122+
List<Point> queryPointsExtract = executor.extractPoints(timePattern, valuePattern);
123+
124+
executor.setPoints(queryPointsExtract);
125+
PatternContext ctx = new PatternContext();
126+
ctx.setThreshold(threshold);
127+
ctx.setDataPoints(sourcePointsExtract);
128+
// State only records time and recorded values, and the final result is calculated
129+
results = executor.executeQuery(ctx);
130+
}
127131
if (!results.isEmpty()) {
128132
resultValue.setBinary(new Binary(results.toString(), Charset.defaultCharset()));
129133
} else {
@@ -140,6 +144,28 @@ public void outputFinal(State state, ResultValue resultValue) {
140144
}
141145
}
142146

147+
private static boolean hasPositiveTimeRange(List<Long> times) {
148+
long previous = times.get(0);
149+
for (int i = 1; i < times.size(); i++) {
150+
long time = times.get(i);
151+
if (time <= previous) {
152+
return false;
153+
}
154+
previous = time;
155+
}
156+
return true;
157+
}
158+
159+
private static boolean hasPositiveValueRange(List<Double> values) {
160+
double min = Double.POSITIVE_INFINITY;
161+
double max = Double.NEGATIVE_INFINITY;
162+
for (double value : values) {
163+
min = Math.min(min, value);
164+
max = Math.max(max, value);
165+
}
166+
return min < max;
167+
}
168+
143169
@Override
144170
public void validate(UDFParameterValidator validator) {
145171

‎library-udf/src/test/java/org/apache/iotdb/library/UDAFPatternTest.java‎

Lines changed: 79 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -214,6 +214,36 @@ public void testDTWMatchValidatesFiniteParameters() {
214214
attributes))));
215215
}
216216

217+
@Test
218+
public void testDTWMatchUsesSinglePointDistance() throws Exception {
219+
UDAFDTWMatch dtwMatch = new UDAFDTWMatch();
220+
UDFParameters parameters = createDtwParameters("10.0", "1");
221+
dtwMatch.validate(new UDFParameterValidator(parameters));
222+
dtwMatch.beforeStart(parameters, new UDAFConfigurations());
223+
DTWState state = (DTWState) dtwMatch.createState();
224+
state.reset();
225+
226+
Column[] columns =
227+
new Column[] {
228+
new TestColumn(TSDataType.DOUBLE, new long[0], new double[] {1.0}, new boolean[] {false}),
229+
new TestColumn(TSDataType.INT64, new long[] {1L}, new double[0], new boolean[] {false})
230+
};
231+
dtwMatch.addInput(state, columns, null);
232+
233+
Assert.assertTrue(state.getMatchResults().isEmpty());
234+
235+
UDAFDTWMatch exactMatch = new UDAFDTWMatch();
236+
UDFParameters exactParameters = createDtwParameters("1.0", "0");
237+
exactMatch.validate(new UDFParameterValidator(exactParameters));
238+
exactMatch.beforeStart(exactParameters, new UDAFConfigurations());
239+
DTWState exactState = (DTWState) exactMatch.createState();
240+
exactState.reset();
241+
242+
exactMatch.addInput(exactState, columns, null);
243+
244+
Assert.assertEquals(1, exactState.getMatchResults().size());
245+
}
246+
217247
@Test
218248
public void testMatchUDAFsSkipNullAndInvalidRows() throws Exception {
219249
Column[] columns = buildPatternInputColumns();
@@ -256,6 +286,41 @@ public void testPatternMatchEmptyEffectiveInputOutputsNull() throws Exception {
256286
Assert.assertTrue(resultBuilder.build().isNull(0));
257287
}
258288

289+
@Test
290+
public void testPatternMatchFlatInputFallsBackToDtw() throws Exception {
291+
UDAFPatternMatch patternMatch = new UDAFPatternMatch();
292+
UDFParameters parameters = createPatternParameters("1,2", "5.0,5.0", "0");
293+
patternMatch.validate(new UDFParameterValidator(parameters));
294+
patternMatch.beforeStart(parameters, new UDAFConfigurations());
295+
PatternState state = (PatternState) patternMatch.createState();
296+
state.reset();
297+
state.updateBuffer(1L, 5.0);
298+
state.updateBuffer(2L, 5.0);
299+
300+
RecordingColumnBuilder resultBuilder = new RecordingColumnBuilder(TSDataType.TEXT);
301+
patternMatch.outputFinal(state, new ResultValue(resultBuilder));
302+
303+
Assert.assertFalse(resultBuilder.build().isNull(0));
304+
}
305+
306+
@Test
307+
public void testPatternMatchNonMonotonicInputFallsBackToDtw() throws Exception {
308+
UDAFPatternMatch patternMatch = new UDAFPatternMatch();
309+
UDFParameters parameters = createPatternParameters("1,2,3", "1.0,2.0,3.0", "0");
310+
patternMatch.validate(new UDFParameterValidator(parameters));
311+
patternMatch.beforeStart(parameters, new UDAFConfigurations());
312+
PatternState state = (PatternState) patternMatch.createState();
313+
state.reset();
314+
state.updateBuffer(1L, 1.0);
315+
state.updateBuffer(3L, 2.0);
316+
state.updateBuffer(2L, 3.0);
317+
318+
RecordingColumnBuilder resultBuilder = new RecordingColumnBuilder(TSDataType.TEXT);
319+
patternMatch.outputFinal(state, new ResultValue(resultBuilder));
320+
321+
Assert.assertFalse(resultBuilder.build().isNull(0));
322+
}
323+
259324
@Test
260325
public void testMatchUDAFsCombineEmptyStates() throws Exception {
261326
UDAFPatternMatch patternMatch = new UDAFPatternMatch();
@@ -305,18 +370,27 @@ public void testMatchUDAFsCombineEmptyStates() throws Exception {
305370
}
306371

307372
private static UDFParameters createPatternParameters() {
373+
return createPatternParameters("1,2", "1.0,2.0", "100");
374+
}
375+
376+
private static UDFParameters createPatternParameters(
377+
String timePattern, String valuePattern, String threshold) {
308378
Map<String, String> attributes = new HashMap<>();
309-
attributes.put("timePattern", "1,2");
310-
attributes.put("valuePattern", "1.0,2.0");
311-
attributes.put("threshold", "100");
379+
attributes.put("timePattern", timePattern);
380+
attributes.put("valuePattern", valuePattern);
381+
attributes.put("threshold", threshold);
312382
return new UDFParameters(
313383
Collections.singletonList("s1"), Collections.singletonList(Type.DOUBLE), attributes);
314384
}
315385

316386
private static UDFParameters createDtwParameters() {
387+
return createDtwParameters("1.0,3.0", "100");
388+
}
389+
390+
private static UDFParameters createDtwParameters(String pattern, String threshold) {
317391
Map<String, String> attributes = new HashMap<>();
318-
attributes.put("pattern", "1.0,3.0");
319-
attributes.put("threshold", "100");
392+
attributes.put("pattern", pattern);
393+
attributes.put("threshold", threshold);
320394
return new UDFParameters(
321395
Collections.singletonList("s1"), Collections.singletonList(Type.DOUBLE), attributes);
322396
}

‎library-udf/src/test/java/org/apache/iotdb/library/UDFWindowAndQueueTest.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1271,6 +1271,45 @@ public void testPatternSymmetrySkipsInvalidWindows() throws Exception {
12711271
Assert.assertTrue(collector.values.isEmpty());
12721272
}
12731273

1274+
@Test
1275+
public void testPatternSymmetryUsesFullWindowDtwCost() throws Exception {
1276+
Map<String, String> attributes = new HashMap<>();
1277+
attributes.put("window", "4");
1278+
attributes.put("threshold", "1650");
1279+
UDFParameters parameters = createSingleDoubleSeriesParameters(attributes);
1280+
UDTFPtnSym ptnSym = new UDTFPtnSym();
1281+
RecordingPointCollector collector = new RecordingPointCollector();
1282+
1283+
ptnSym.validate(new UDFParameterValidator(parameters));
1284+
ptnSym.beforeStart(parameters, new UDTFConfigurations(ZoneId.systemDefault()));
1285+
ptnSym.transform(
1286+
new SimpleRowWindow(
1287+
new DoubleRow(1, 0.0),
1288+
new DoubleRow(2, 10.0),
1289+
new DoubleRow(3, 20.0),
1290+
new DoubleRow(4, 40.0)),
1291+
collector);
1292+
1293+
Assert.assertTrue(collector.timestamps.isEmpty());
1294+
Assert.assertTrue(collector.values.isEmpty());
1295+
1296+
attributes.put("threshold", "0");
1297+
parameters = createSingleDoubleSeriesParameters(attributes);
1298+
ptnSym = new UDTFPtnSym();
1299+
ptnSym.validate(new UDFParameterValidator(parameters));
1300+
ptnSym.beforeStart(parameters, new UDTFConfigurations(ZoneId.systemDefault()));
1301+
ptnSym.transform(
1302+
new SimpleRowWindow(
1303+
new DoubleRow(5, 1.0),
1304+
new DoubleRow(6, 2.0),
1305+
new DoubleRow(7, 2.0),
1306+
new DoubleRow(8, 1.0)),
1307+
collector);
1308+
1309+
Assert.assertEquals(Collections.singletonList(5L), collector.timestamps);
1310+
Assert.assertEquals(0.0, collector.values.get(0), 0.0);
1311+
}
1312+
12741313
@Test
12751314
public void testDtwEmptyInputProducesNoOutput() throws Exception {
12761315
UDAFDtw dtw = new UDAFDtw();

0 commit comments

Comments
 (0)