Skip to content

Commit b885269

Browse files
CaideyipiCaideyipi
authored
Fix operator tests for transient null TsBlocks (#17838) (#17847)
1 parent 0288aab commit b885269

10 files changed

Lines changed: 118 additions & 48 deletions

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/AlignedSeriesAggregationScanOperatorTest.java‎

Lines changed: 17 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@
6969
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
7070
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationOperatorTest.TEST_TIME_SLICE;
7171
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
72+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
7273
import static org.junit.Assert.assertEquals;
7374
import static org.junit.Assert.assertTrue;
7475

@@ -122,7 +123,7 @@ public void testAggregationWithoutTimeFilter() throws Exception {
122123
initAlignedSeriesAggregationScanOperator(aggregators, null, true, null);
123124
int count = 0;
124125
while (seriesAggregationScanOperator.hasNext()) {
125-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
126+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
126127
for (int i = 0; i < measurementSchemas.size(); i++) {
127128
assertEquals(500, resultTsBlock.getColumn(i).getLong(0));
128129
}
@@ -153,7 +154,7 @@ public void testAggregationWithoutTimeFilterOrderByTimeDesc() throws Exception {
153154
initAlignedSeriesAggregationScanOperator(aggregators, null, false, null);
154155
int count = 0;
155156
while (seriesAggregationScanOperator.hasNext()) {
156-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
157+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
157158
for (int i = 0; i < measurementSchemas.size(); i++) {
158159
assertEquals(500, resultTsBlock.getColumn(i).getLong(0));
159160
}
@@ -189,7 +190,7 @@ public void testMultiAggregationFuncWithoutTimeFilter1() throws Exception {
189190
initAlignedSeriesAggregationScanOperator(aggregators, null, true, null);
190191
int count = 0;
191192
while (seriesAggregationScanOperator.hasNext()) {
192-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
193+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
193194
assertEquals(500, resultTsBlock.getColumn(0).getLong(0));
194195
assertEquals(6524750.0, resultTsBlock.getColumn(1).getDouble(0), 0.0001);
195196
count++;
@@ -227,7 +228,7 @@ public void testMultiAggregationFuncWithoutTimeFilter2() throws Exception {
227228
initAlignedSeriesAggregationScanOperator(aggregators, null, true, null);
228229
int count = 0;
229230
while (seriesAggregationScanOperator.hasNext()) {
230-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
231+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
231232
assertTrue(resultTsBlock.getColumn(0).getBoolean(0));
232233
assertEquals(10499, resultTsBlock.getColumn(1).getInt(0));
233234
assertEquals(20199, resultTsBlock.getColumn(2).getLong(0));
@@ -269,7 +270,7 @@ public void testMultiAggregationFuncWithoutTimeFilterOrderByTimeDesc() throws Ex
269270
initAlignedSeriesAggregationScanOperator(aggregators, null, false, null);
270271
int count = 0;
271272
while (seriesAggregationScanOperator.hasNext()) {
272-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
273+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
273274
assertTrue(resultTsBlock.getColumn(0).getBoolean(0));
274275
assertEquals(10499, resultTsBlock.getColumn(1).getInt(0));
275276
assertEquals(20199, resultTsBlock.getColumn(2).getLong(0));
@@ -304,7 +305,7 @@ public void testAggregationWithTimeFilter1() throws Exception {
304305
initAlignedSeriesAggregationScanOperator(aggregators, timeFilter, true, null);
305306
int count = 0;
306307
while (seriesAggregationScanOperator.hasNext()) {
307-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
308+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
308309
for (int i = 0; i < measurementSchemas.size(); i++) {
309310
assertEquals(resultTsBlock.getColumn(i).getLong(0), 380);
310311
}
@@ -337,7 +338,7 @@ public void testAggregationWithTimeFilter2() throws Exception {
337338
initAlignedSeriesAggregationScanOperator(aggregators, timeFilter, true, null);
338339
int count = 0;
339340
while (seriesAggregationScanOperator.hasNext()) {
340-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
341+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
341342
for (int i = 0; i < measurementSchemas.size(); i++) {
342343
assertEquals(resultTsBlock.getColumn(i).getLong(0), 380);
343344
}
@@ -370,7 +371,7 @@ public void testAggregationWithTimeFilter3() throws Exception {
370371
initAlignedSeriesAggregationScanOperator(aggregators, timeFilter, true, null);
371372
int count = 0;
372373
while (seriesAggregationScanOperator.hasNext()) {
373-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
374+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
374375
for (int i = 0; i < measurementSchemas.size(); i++) {
375376
assertEquals(resultTsBlock.getColumn(i).getLong(0), 300);
376377
}
@@ -410,7 +411,7 @@ public void testMultiAggregationWithTimeFilter() throws Exception {
410411
initAlignedSeriesAggregationScanOperator(aggregators, timeFilter, true, null);
411412
int count = 0;
412413
while (seriesAggregationScanOperator.hasNext()) {
413-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
414+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
414415
assertTrue(resultTsBlock.getColumn(0).getBoolean(0));
415416
assertEquals(399, resultTsBlock.getColumn(1).getInt(0));
416417
assertEquals(20199, resultTsBlock.getColumn(2).getLong(0));
@@ -448,7 +449,7 @@ public void testGroupByWithoutGlobalTimeFilter() throws Exception {
448449
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
449450
int count = 0;
450451
while (seriesAggregationScanOperator.hasNext()) {
451-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
452+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
452453
int positionCount = resultTsBlock.getPositionCount();
453454
for (int pos = 0; pos < positionCount; pos++) {
454455
assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -489,7 +490,7 @@ public void testGroupByWithGlobalTimeFilter() throws Exception {
489490
aggregators, timeFilter, true, groupByTimeParameter);
490491
int count = 0;
491492
while (seriesAggregationScanOperator.hasNext()) {
492-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
493+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
493494
int positionCount = resultTsBlock.getPositionCount();
494495
for (int pos = 0; pos < positionCount; pos++) {
495496
assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -533,7 +534,7 @@ public void testGroupByWithMultiFunction() throws Exception {
533534
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
534535
int count = 0;
535536
while (seriesAggregationScanOperator.hasNext()) {
536-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
537+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
537538
int positionCount = resultTsBlock.getPositionCount();
538539
for (int pos = 0; pos < positionCount; pos++) {
539540
assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -578,7 +579,7 @@ public void testGroupByWithMultiFunctionOrderByTimeDesc() throws Exception {
578579
initAlignedSeriesAggregationScanOperator(aggregators, null, false, groupByTimeParameter);
579580
int count = 0;
580581
while (seriesAggregationScanOperator.hasNext()) {
581-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
582+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
582583
int positionCount = resultTsBlock.getPositionCount();
583584
for (int pos = 0; pos < positionCount; pos++) {
584585
assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(pos));
@@ -613,7 +614,7 @@ public void testGroupBySlidingTimeWindow() throws Exception {
613614
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
614615
int count = 0;
615616
while (seriesAggregationScanOperator.hasNext()) {
616-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
617+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
617618
int positionCount = resultTsBlock.getPositionCount();
618619
for (int pos = 0; pos < positionCount; pos++) {
619620
assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -645,7 +646,7 @@ public void testGroupBySlidingTimeWindow2() throws Exception {
645646
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
646647
int count = 0;
647648
while (seriesAggregationScanOperator.hasNext()) {
648-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
649+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
649650
int positionCount = resultTsBlock.getPositionCount();
650651
for (int pos = 0; pos < positionCount; pos++) {
651652
assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
@@ -688,7 +689,7 @@ public void testGroupBySlidingWindowWithMultiFunction() throws Exception {
688689
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
689690
int count = 0;
690691
while (seriesAggregationScanOperator.hasNext()) {
691-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
692+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
692693
int positionCount = resultTsBlock.getPositionCount();
693694
for (int pos = 0; pos < positionCount; pos++) {
694695
assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/FillOperatorTest.java‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
import java.util.concurrent.ExecutorService;
4343

4444
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
45+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
4546
import static org.junit.Assert.assertEquals;
4647
import static org.junit.Assert.assertTrue;
4748

@@ -209,7 +210,7 @@ public long ramBytesUsed() {
209210
}
210211
};
211212
while (fillOperator.hasNext()) {
212-
TsBlock block = fillOperator.next();
213+
TsBlock block = nextNonNullOrEmpty(fillOperator);
213214
for (int i = 0; i < block.getPositionCount(); i++) {
214215
long expectedTime = i + 1 + count * 10000L;
215216
assertEquals(expectedTime, block.getTimeByIndex(i));
@@ -386,7 +387,7 @@ public long ramBytesUsed() {
386387
}
387388
};
388389
while (fillOperator.hasNext()) {
389-
TsBlock block = fillOperator.next();
390+
TsBlock block = nextNonNullOrEmpty(fillOperator);
390391
for (int i = 0; i < block.getPositionCount(); i++) {
391392
long expectedTime = i + 1 + count * 10000L;
392393
assertEquals(expectedTime, block.getTimeByIndex(i));
@@ -563,7 +564,7 @@ public long ramBytesUsed() {
563564
}
564565
};
565566
while (fillOperator.hasNext()) {
566-
TsBlock block = fillOperator.next();
567+
TsBlock block = nextNonNullOrEmpty(fillOperator);
567568
for (int i = 0; i < block.getPositionCount(); i++) {
568569
long expectedTime = i + 1 + count * 10000L;
569570
assertEquals(expectedTime, block.getTimeByIndex(i));

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/HorizontallyConcatOperatorTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@
6666

6767
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
6868
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
69+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
6970
import static org.junit.Assert.assertEquals;
7071
import static org.junit.Assert.fail;
7172

@@ -197,7 +198,7 @@ public void batchTest1() throws Exception {
197198
int count = 0;
198199
while (horizontallyConcatOperator.isBlocked().isDone()
199200
&& horizontallyConcatOperator.hasNext()) {
200-
TsBlock tsBlock = horizontallyConcatOperator.next();
201+
TsBlock tsBlock = nextNonNullOrEmpty(horizontallyConcatOperator);
201202
assertEquals(6, tsBlock.getValueColumnCount());
202203
for (int i = 0; i < tsBlock.getPositionCount(); i++, count++) {
203204
assertEquals(count, tsBlock.getTimeByIndex(i));

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LinearFillOperatorTest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -41,8 +41,8 @@
4141
import java.util.concurrent.ExecutorService;
4242

4343
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
44+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
4445
import static org.junit.Assert.assertEquals;
45-
import static org.junit.Assert.assertNotNull;
4646
import static org.junit.Assert.assertTrue;
4747

4848
public class LinearFillOperatorTest {
@@ -1301,8 +1301,7 @@ public long ramBytesUsed() {
13011301
};
13021302

13031303
while (fillOperator.hasNext()) {
1304-
TsBlock block = fillOperator.next();
1305-
assertNotNull(block);
1304+
TsBlock block = nextNonNullOrEmpty(fillOperator);
13061305
for (int i = 0; i < block.getPositionCount(); i++) {
13071306
long expectedTime = i + count;
13081307
assertEquals(expectedTime, block.getTimeByIndex(i));

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@
8888
import java.util.concurrent.atomic.AtomicInteger;
8989

9090
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
91+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
9192
import static org.junit.Assert.assertEquals;
9293
import static org.junit.Assert.assertTrue;
9394
import static org.junit.Assert.fail;
@@ -1765,7 +1766,7 @@ public void mergeSortTest() throws Exception {
17651766

17661767
int index = 0;
17671768
while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
1768-
TsBlock result = mergeSortOperator.next();
1769+
TsBlock result = nextNonNullOrEmpty(mergeSortOperator);
17691770
for (int i = 0; i < result.getPositionCount(); i++) {
17701771
long time = result.getTimeByIndex(i);
17711772
assertEquals(time, ans[index++]);

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OffsetOperatorTest.java‎

Lines changed: 15 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -62,6 +62,7 @@
6262
import java.util.concurrent.TimeUnit;
6363

6464
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
65+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
6566
import static org.junit.Assert.assertEquals;
6667
import static org.junit.Assert.assertTrue;
6768
import static org.junit.Assert.fail;
@@ -161,7 +162,7 @@ public void batchTest1() throws Exception {
161162
new LimitOperator(driverContext.getOperatorContexts().get(4), 250, offsetOperator);
162163
int count = 100;
163164
while (limitOperator.isBlocked().isDone() && limitOperator.hasNext()) {
164-
TsBlock tsBlock = limitOperator.next();
165+
TsBlock tsBlock = nextNonNullOrEmpty(limitOperator);
165166
assertEquals(2, tsBlock.getValueColumnCount());
166167
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
167168
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -265,7 +266,7 @@ public void batchTest2() throws Exception {
265266

266267
int count = 0;
267268
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
268-
TsBlock tsBlock = offsetOperator.next();
269+
TsBlock tsBlock = nextNonNullOrEmpty(offsetOperator);
269270
assertEquals(2, tsBlock.getValueColumnCount());
270271
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
271272
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -367,7 +368,7 @@ public void batchTest3() throws Exception {
367368
new OffsetOperator(driverContext.getOperatorContexts().get(3), 500, timeJoinOperator);
368369

369370
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
370-
TsBlock tsBlock = offsetOperator.next();
371+
TsBlock tsBlock = nextNonNull(offsetOperator);
371372
assertEquals(2, tsBlock.getValueColumnCount());
372373
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
373374
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -455,7 +456,7 @@ public void batchTest4() throws Exception {
455456
driverContext.getOperatorContexts().get(3), 98_784_247_808L, timeJoinOperator);
456457

457458
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
458-
TsBlock tsBlock = offsetOperator.next();
459+
TsBlock tsBlock = nextNonNull(offsetOperator);
459460
assertEquals(2, tsBlock.getValueColumnCount());
460461
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
461462
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -468,4 +469,14 @@ public void batchTest4() throws Exception {
468469
instanceNotificationExecutor.shutdown();
469470
}
470471
}
472+
473+
private static TsBlock nextNonNull(Operator operator) throws Exception {
474+
while (operator.hasNext()) {
475+
TsBlock result = operator.next();
476+
if (result != null) {
477+
return result;
478+
}
479+
}
480+
throw new AssertionError("Expected a non-null TsBlock from operator");
481+
}
471482
}

‎iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorTestUtils.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.iotdb.db.queryengine.execution.operator;
20+
21+
import org.apache.tsfile.read.common.block.TsBlock;
22+
23+
public final class OperatorTestUtils {
24+
25+
private OperatorTestUtils() {
26+
// Utility class.
27+
}
28+
29+
public static TsBlock nextNonNullOrEmpty(Operator operator) throws Exception {
30+
while (operator.hasNext()) {
31+
TsBlock result = operator.next();
32+
if (!isNullOrEmpty(result)) {
33+
return result;
34+
}
35+
}
36+
throw new AssertionError("Expected a non-null and non-empty TsBlock from operator");
37+
}
38+
39+
public static TsBlock lastNonNullOrEmpty(Operator operator) throws Exception {
40+
TsBlock result = null;
41+
while (operator.isBlocked().isDone() && operator.hasNext()) {
42+
TsBlock nextResult = operator.next();
43+
if (!isNullOrEmpty(nextResult)) {
44+
result = nextResult;
45+
}
46+
}
47+
return result;
48+
}
49+
50+
private static boolean isNullOrEmpty(TsBlock tsBlock) {
51+
return tsBlock == null || tsBlock.getPositionCount() == 0;
52+
}
53+
}

0 commit comments

Comments
 (0)