Skip to content

Commit 16d3669

Browse files
CaideyipiCaideyipi
authored
Fix operator tests for transient null TsBlocks (#17838)
1 parent 373af66 commit 16d3669

12 files changed

Lines changed: 127 additions & 64 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));
@@ -534,7 +535,7 @@ public void testGroupByWithMultiFunction() throws Exception {
534535
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
535536
int count = 0;
536537
while (seriesAggregationScanOperator.hasNext()) {
537-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
538+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
538539
int positionCount = resultTsBlock.getPositionCount();
539540
for (int pos = 0; pos < positionCount; pos++) {
540541
assertEquals(100 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -580,7 +581,7 @@ public void testGroupByWithMultiFunctionOrderByTimeDesc() throws Exception {
580581
initAlignedSeriesAggregationScanOperator(aggregators, null, false, groupByTimeParameter);
581582
int count = 0;
582583
while (seriesAggregationScanOperator.hasNext()) {
583-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
584+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
584585
int positionCount = resultTsBlock.getPositionCount();
585586
for (int pos = 0; pos < positionCount; pos++) {
586587
assertEquals(100 * (3 - count), resultTsBlock.getTimeColumn().getLong(pos));
@@ -616,7 +617,7 @@ public void testGroupBySlidingTimeWindow() throws Exception {
616617
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
617618
int count = 0;
618619
while (seriesAggregationScanOperator.hasNext()) {
619-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
620+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
620621
int positionCount = resultTsBlock.getPositionCount();
621622
for (int pos = 0; pos < positionCount; pos++) {
622623
assertEquals(50 * count, resultTsBlock.getTimeColumn().getLong(pos));
@@ -649,7 +650,7 @@ public void testGroupBySlidingTimeWindow2() throws Exception {
649650
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
650651
int count = 0;
651652
while (seriesAggregationScanOperator.hasNext()) {
652-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
653+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
653654
int positionCount = resultTsBlock.getPositionCount();
654655
for (int pos = 0; pos < positionCount; pos++) {
655656
assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));
@@ -693,7 +694,7 @@ public void testGroupBySlidingWindowWithMultiFunction() throws Exception {
693694
initAlignedSeriesAggregationScanOperator(aggregators, null, true, groupByTimeParameter);
694695
int count = 0;
695696
while (seriesAggregationScanOperator.hasNext()) {
696-
TsBlock resultTsBlock = seriesAggregationScanOperator.next();
697+
TsBlock resultTsBlock = nextNonNullOrEmpty(seriesAggregationScanOperator);
697698
int positionCount = resultTsBlock.getPositionCount();
698699
for (int pos = 0; pos < positionCount; pos++) {
699700
assertEquals(timeColumn[count], resultTsBlock.getTimeColumn().getLong(pos));

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

Lines changed: 4 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@
7272

7373
import static com.google.common.util.concurrent.Futures.immediateFuture;
7474
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
75+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.lastNonNullOrEmpty;
7576
import static org.apache.iotdb.rpc.RpcUtils.SUCCESS_STATUS;
7677
import static org.junit.Assert.assertEquals;
7778
import static org.junit.Assert.assertNotNull;
@@ -258,10 +259,7 @@ public void testSingleDeviceSmallData() throws Exception {
258259
prepareDeviceData("device0", 2);
259260
operator = createAndInitOperatorForSingleDevices(2);
260261

261-
TsBlock result = null;
262-
while (operator.isBlocked().isDone() && operator.hasNext()) {
263-
result = operator.next();
264-
}
262+
TsBlock result = lastNonNullOrEmpty(operator);
265263
assertNotNull(result);
266264
assertEquals(2, result.getPositionCount());
267265

@@ -276,10 +274,7 @@ public void testSingleDeviceExceedsMaxTsBlockSize() throws Exception {
276274
prepareDeviceData("device0", 10);
277275
operator = createAndInitOperatorForSingleDevices(10);
278276

279-
TsBlock result = null;
280-
while (operator.isBlocked().isDone() && operator.hasNext()) {
281-
result = operator.next();
282-
}
277+
TsBlock result = lastNonNullOrEmpty(operator);
283278
assertNotNull(result);
284279
assertEquals(10, result.getPositionCount());
285280

@@ -295,10 +290,7 @@ public void testMultipleDeviceSmallData() throws Exception {
295290
prepareDeviceData("device1", 1);
296291
operator = createAndInitOperatorForMultipleDevices(2, 1);
297292

298-
TsBlock result = null;
299-
while (operator.isBlocked().isDone() && operator.hasNext()) {
300-
result = operator.next();
301-
}
293+
TsBlock result = lastNonNullOrEmpty(operator);
302294
assertNotNull(result);
303295
assertEquals(2, result.getPositionCount());
304296

‎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
@@ -43,6 +43,7 @@
4343
import java.util.concurrent.ExecutorService;
4444

4545
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
46+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
4647
import static org.junit.Assert.assertEquals;
4748
import static org.junit.Assert.assertTrue;
4849

@@ -210,7 +211,7 @@ public long ramBytesUsed() {
210211
}
211212
};
212213
while (fillOperator.hasNext()) {
213-
TsBlock block = fillOperator.next();
214+
TsBlock block = nextNonNullOrEmpty(fillOperator);
214215
for (int i = 0; i < block.getPositionCount(); i++) {
215216
long expectedTime = i + 1 + count * 10000L;
216217
assertEquals(expectedTime, block.getTimeByIndex(i));
@@ -383,7 +384,7 @@ public long ramBytesUsed() {
383384
}
384385
};
385386
while (fillOperator.hasNext()) {
386-
TsBlock block = fillOperator.next();
387+
TsBlock block = nextNonNullOrEmpty(fillOperator);
387388
for (int i = 0; i < block.getPositionCount(); i++) {
388389
long expectedTime = i + 1 + count * 10000L;
389390
assertEquals(expectedTime, block.getTimeByIndex(i));
@@ -560,7 +561,7 @@ public long ramBytesUsed() {
560561
}
561562
};
562563
while (fillOperator.hasNext()) {
563-
TsBlock block = fillOperator.next();
564+
TsBlock block = nextNonNullOrEmpty(fillOperator);
564565
for (int i = 0; i < block.getPositionCount(); i++) {
565566
long expectedTime = i + 1 + count * 10000L;
566567
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
@@ -68,6 +68,7 @@
6868

6969
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
7070
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.initTimeRangeIterator;
71+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
7172
import static org.junit.Assert.assertEquals;
7273
import static org.junit.Assert.fail;
7374

@@ -203,7 +204,7 @@ public void batchTest1() throws Exception {
203204
int count = 0;
204205
while (horizontallyConcatOperator.isBlocked().isDone()
205206
&& horizontallyConcatOperator.hasNext()) {
206-
TsBlock tsBlock = horizontallyConcatOperator.next();
207+
TsBlock tsBlock = nextNonNullOrEmpty(horizontallyConcatOperator);
207208
assertEquals(6, tsBlock.getValueColumnCount());
208209
for (int i = 0; i < tsBlock.getPositionCount(); i++, count++) {
209210
assertEquals(count, tsBlock.getTimeByIndex(i));

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

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

9595
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
96+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
9697
import static org.junit.Assert.assertEquals;
9798
import static org.junit.Assert.assertTrue;
9899
import static org.junit.Assert.fail;
@@ -1794,7 +1795,7 @@ public void mergeSortTest() throws Exception {
17941795

17951796
int index = 0;
17961797
while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
1797-
TsBlock result = treeMergeSortOperator.next();
1798+
TsBlock result = nextNonNullOrEmpty(treeMergeSortOperator);
17981799
for (int i = 0; i < result.getPositionCount(); i++) {
17991800
long time = result.getTimeByIndex(i);
18001801
assertEquals(time, ans[index++]);

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

Lines changed: 16 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
*/
1919
package org.apache.iotdb.db.queryengine.execution.operator;
2020

21+
import org.apache.iotdb.calc.execution.operator.Operator;
2122
import org.apache.iotdb.calc.execution.operator.process.LimitOperator;
2223
import org.apache.iotdb.calc.execution.operator.process.OffsetOperator;
2324
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
@@ -65,6 +66,7 @@
6566
import java.util.concurrent.TimeUnit;
6667

6768
import static org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
69+
import static org.apache.iotdb.db.queryengine.execution.operator.OperatorTestUtils.nextNonNullOrEmpty;
6870
import static org.junit.Assert.assertEquals;
6971
import static org.junit.Assert.assertTrue;
7072
import static org.junit.Assert.fail;
@@ -168,7 +170,7 @@ public void batchTest1() throws Exception {
168170
new LimitOperator(driverContext.getOperatorContexts().get(4), 250, offsetOperator);
169171
int count = 100;
170172
while (limitOperator.isBlocked().isDone() && limitOperator.hasNext()) {
171-
TsBlock tsBlock = limitOperator.next();
173+
TsBlock tsBlock = nextNonNullOrEmpty(limitOperator);
172174
assertEquals(2, tsBlock.getValueColumnCount());
173175
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
174176
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -276,7 +278,7 @@ public void batchTest2() throws Exception {
276278

277279
int count = 0;
278280
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
279-
TsBlock tsBlock = offsetOperator.next();
281+
TsBlock tsBlock = nextNonNullOrEmpty(offsetOperator);
280282
assertEquals(2, tsBlock.getValueColumnCount());
281283
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
282284
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -382,7 +384,7 @@ public void batchTest3() throws Exception {
382384
new OffsetOperator(driverContext.getOperatorContexts().get(3), 500, timeJoinOperator);
383385

384386
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
385-
TsBlock tsBlock = offsetOperator.next();
387+
TsBlock tsBlock = nextNonNull(offsetOperator);
386388
assertEquals(2, tsBlock.getValueColumnCount());
387389
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
388390
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -474,7 +476,7 @@ public void batchTest4() throws Exception {
474476
driverContext.getOperatorContexts().get(3), 98_784_247_808L, timeJoinOperator);
475477

476478
while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
477-
TsBlock tsBlock = offsetOperator.next();
479+
TsBlock tsBlock = nextNonNull(offsetOperator);
478480
assertEquals(2, tsBlock.getValueColumnCount());
479481
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
480482
assertTrue(tsBlock.getColumn(1) instanceof IntColumn);
@@ -487,4 +489,14 @@ public void batchTest4() throws Exception {
487489
instanceNotificationExecutor.shutdown();
488490
}
489491
}
492+
493+
private static TsBlock nextNonNull(Operator operator) throws Exception {
494+
while (operator.hasNext()) {
495+
TsBlock result = operator.next();
496+
if (result != null) {
497+
return result;
498+
}
499+
}
500+
throw new AssertionError("Expected a non-null TsBlock from operator");
501+
}
490502
}

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

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,55 @@
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.iotdb.calc.execution.operator.Operator;
22+
23+
import org.apache.tsfile.read.common.block.TsBlock;
24+
25+
public final class OperatorTestUtils {
26+
27+
private OperatorTestUtils() {
28+
// Utility class.
29+
}
30+
31+
public static TsBlock nextNonNullOrEmpty(Operator operator) throws Exception {
32+
while (operator.hasNext()) {
33+
TsBlock result = operator.next();
34+
if (!isNullOrEmpty(result)) {
35+
return result;
36+
}
37+
}
38+
throw new AssertionError("Expected a non-null and non-empty TsBlock from operator");
39+
}
40+
41+
public static TsBlock lastNonNullOrEmpty(Operator operator) throws Exception {
42+
TsBlock result = null;
43+
while (operator.isBlocked().isDone() && operator.hasNext()) {
44+
TsBlock nextResult = operator.next();
45+
if (!isNullOrEmpty(nextResult)) {
46+
result = nextResult;
47+
}
48+
}
49+
return result;
50+
}
51+
52+
private static boolean isNullOrEmpty(TsBlock tsBlock) {
53+
return tsBlock == null || tsBlock.getPositionCount() == 0;
54+
}
55+
}

0 commit comments

Comments
 (0)