You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ca...@apache.org on 2023/10/10 06:20:23 UTC
[iotdb] branch beyyes/fix_order_by_limit updated: add test
This is an automated email from the ASF dual-hosted git repository.
caogaofei pushed a commit to branch beyyes/fix_order_by_limit
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/beyyes/fix_order_by_limit by this push:
new 023d1ef865e add test
023d1ef865e is described below
commit 023d1ef865ee851571965d80e4224013ffb76cf0
Author: Beyyes <cg...@foxmail.com>
AuthorDate: Tue Oct 10 14:20:06 2023 +0800
add test
---
.../execution/operator/process/TopKOperator.java | 157 ++---
.../execution/operator/MergeSortOperatorTest.java | 81 +--
...SortOperatorTest.java => TopKOperatorTest.java} | 754 +++------------------
3 files changed, 223 insertions(+), 769 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
index cf5eedc505d..97ce809fdbc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TopKOperator.java
@@ -52,7 +52,7 @@ public class TopKOperator extends AbstractConsumeAllOperator {
private final long topValue;
- private int readingValueCount = 0;
+ boolean allBlocksRead;
public TopKOperator(
OperatorContext operatorContext,
@@ -63,7 +63,7 @@ public class TopKOperator extends AbstractConsumeAllOperator {
super(operatorContext, inputOperators);
this.dataTypes = dataTypes;
// use MAX-HEAP
- this.mergeSortHeap = new MergeSortHeap(inputOperatorsCount, comparator.reversed());
+ this.mergeSortHeap = new MergeSortHeap((int) topValue, comparator.reversed());
this.comparator = comparator;
this.noMoreTsBlocks = new boolean[inputOperatorsCount];
this.tsBlockBuilder = new TsBlockBuilder((int) topValue, dataTypes);
@@ -82,34 +82,12 @@ public class TopKOperator extends AbstractConsumeAllOperator {
listenableFutures.add(blocked);
}
}
- return listenableFutures.isEmpty()
- ? NOT_BLOCKED
- : successfulAsList(listenableFutures);
+ return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
}
@Override
public boolean hasNext() throws Exception {
- if (finished) {
- return false;
- }
- if (readingValueCount > topValue) {
- return false;
- }
- for (int i = 0; i < inputOperatorsCount; i++) {
- if (!isEmpty(i)) {
- return true;
- } else if (!noMoreTsBlocks[i]) {
- if (!canCallNext[i] || children.get(i).hasNextWithTimer()) {
- return true;
- } else {
- children.get(i).close();
- children.set(i, null);
- noMoreTsBlocks[i] = true;
- inputTsBlocks[i] = null;
- }
- }
- }
- return false;
+ return !allBlocksRead;
}
@Override
@@ -117,30 +95,48 @@ public class TopKOperator extends AbstractConsumeAllOperator {
long startTime = System.nanoTime();
long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
- // 1. fill consumed up TsBlock
- if (!prepareInput()) {
- return null;
- }
-
- boolean allBlocksRead = true;
+ allBlocksRead = true;
for (int i = 0; i < inputOperatorsCount; i++) {
if (noMoreTsBlocks[i] || children.get(i) == null) {
continue;
}
- allBlocksRead = false;
- boolean getNewValue = false;
- for (int idx = 0; idx < inputTsBlocks[i].getPositionCount(); idx++) {
- if (mergeSortHeap.getHeapSize() < topValue || comparator.compare(new MergeSortKey(inputTsBlocks[i], idx),
- mergeSortHeap.peek()) < 0) {
- mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], idx));
- getNewValue = true;
- } else {
- if (!getNewValue) {
- inputTsBlocks[i] = null;
- noMoreTsBlocks[i] = true;
+ if (!children.get(i).isBlocked().isDone()) {
+ allBlocksRead = false;
+ continue;
+ }
+
+ if (children.get(i).hasNextWithTimer()) {
+ inputTsBlocks[i] = getNextTsBlock(i);
+ if (inputTsBlocks[i] == null) {
+ allBlocksRead = false;
+ continue;
+ }
+
+ boolean getNewValue = false;
+ for (int idx = 0; idx < inputTsBlocks[i].getPositionCount(); idx++) {
+ if (mergeSortHeap.getHeapSize() < topValue) {
+ mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], idx));
+ getNewValue = true;
+ } else if (comparator.compare(
+ new MergeSortKey(inputTsBlocks[i], idx), mergeSortHeap.peek())
+ < 0) {
+ mergeSortHeap.poll();
+ mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], idx));
+ getNewValue = true;
}
}
+
+ if (!getNewValue) {
+ inputTsBlocks[i] = null;
+ noMoreTsBlocks[i] = true;
+ children.set(i, null);
+ } else {
+ allBlocksRead = false;
+ }
+ } else {
+ noMoreTsBlocks[i] = true;
+ children.set(i, null);
}
}
@@ -148,31 +144,13 @@ public class TopKOperator extends AbstractConsumeAllOperator {
return null;
}
- // 3. do merge sort until one TsBlock is consumed up
- tsBlockBuilder.reset();
- TimeColumnBuilder timeBuilder = tsBlockBuilder.getTimeColumnBuilder();
- ColumnBuilder[] valueColumnBuilders = tsBlockBuilder.getValueColumnBuilders();
- while (!mergeSortHeap.isEmpty()) {
- MergeSortKey mergeSortKey = mergeSortHeap.poll();
- TsBlock targetBlock = mergeSortKey.tsBlock;
- int rowIndex = mergeSortKey.rowIndex;
- timeBuilder.writeLong(targetBlock.getTimeByIndex(rowIndex));
- for (int i = 0; i < valueColumnBuilders.length; i++) {
- if (targetBlock.getColumn(i).isNull(rowIndex)) {
- valueColumnBuilders[i].appendNull();
- continue;
- }
- valueColumnBuilders[i].write(targetBlock.getColumn(i), rowIndex);
- }
- tsBlockBuilder.declarePosition();
-
- // break if time is out or tsBlockBuilder is full
- if (System.nanoTime() - startTime > maxRuntime) {
- return null;
- }
+ // break if time is out
+ // when tsBlockBuilder is full but the mergeSortHeap is still not empty, we also return null
+ if (System.nanoTime() - startTime > maxRuntime) {
+ return null;
}
- return tsBlockBuilder.build();
+ return getResultFromMaxHeap(mergeSortHeap, tsBlockBuilder);
}
@Override
@@ -206,30 +184,33 @@ public class TopKOperator extends AbstractConsumeAllOperator {
return 0;
}
- @Override
- protected boolean prepareInput() throws Exception {
- boolean allReady = true;
- for (int i = 0; i < inputOperatorsCount; i++) {
- if (noMoreTsBlocks[i] || !isEmpty(i) || children.get(i) == null) {
- continue;
- }
- if (canCallNext[i]) {
- if (children.get(i).hasNextWithTimer()) {
- inputTsBlocks[i] = getNextTsBlock(i);
- canCallNext[i] = false;
- if (isEmpty(i)) {
- allReady = false;
- } else {
- mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], 0, i));
- }
- } else {
- noMoreTsBlocks[i] = true;
- inputTsBlocks[i] = null;
+ private TsBlock getResultFromMaxHeap(MergeSortHeap mergeSortHeap, TsBlockBuilder tsBlockBuilder) {
+ int cnt = mergeSortHeap.getHeapSize();
+ MergeSortKey[] mergeSortKeys = new MergeSortKey[cnt];
+ while (!mergeSortHeap.isEmpty()) {
+ mergeSortKeys[--cnt] = mergeSortHeap.poll();
+ }
+
+ tsBlockBuilder.reset();
+ TimeColumnBuilder timeBuilder = tsBlockBuilder.getTimeColumnBuilder();
+ ColumnBuilder[] valueColumnBuilders = tsBlockBuilder.getValueColumnBuilders();
+ for (MergeSortKey mergeSortKey : mergeSortKeys) {
+ TsBlock targetBlock = mergeSortKey.tsBlock;
+ int rowIndex = mergeSortKey.rowIndex;
+ timeBuilder.writeLong(targetBlock.getTimeByIndex(rowIndex));
+ for (int i = 0; i < valueColumnBuilders.length; i++) {
+ if (targetBlock.getColumn(i).isNull(rowIndex)) {
+ valueColumnBuilders[i].appendNull();
+ continue;
}
- } else {
- allReady = false;
+ valueColumnBuilders[i].write(targetBlock.getColumn(i), rowIndex);
}
+ tsBlockBuilder.declarePosition();
}
- return allReady;
+
+ allBlocksRead = true;
+ finished = true;
+ // TODO make all children null
+ return tsBlockBuilder.build();
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
index 3d7e6346981..08f66af328a 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
@@ -362,7 +362,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -370,7 +370,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -378,7 +378,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -386,7 +386,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -409,7 +409,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -417,7 +417,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
@@ -426,7 +426,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -434,7 +434,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -457,7 +457,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -465,7 +465,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
@@ -474,7 +474,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -482,7 +482,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
@Test
@@ -506,7 +506,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -514,7 +514,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -522,7 +522,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -530,7 +530,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(1500, count);
}
// ------------------------------------------------------------------------------------------------
@@ -856,22 +856,22 @@ public class MergeSortOperatorTest {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -879,7 +879,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -900,23 +900,23 @@ public class MergeSortOperatorTest {
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -924,7 +924,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -944,23 +944,23 @@ public class MergeSortOperatorTest {
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -968,7 +968,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -989,22 +989,22 @@ public class MergeSortOperatorTest {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -1012,8 +1012,9 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
+
// ------------------------------------------------------------------------------------------------
// order by device
// ------------------------------------------------------------------------------------------------
@@ -1332,7 +1333,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1385,7 +1386,7 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1438,7 +1439,7 @@ public class MergeSortOperatorTest {
}
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
@Test
@@ -1490,7 +1491,7 @@ public class MergeSortOperatorTest {
}
}
}
- assertEquals(count, 2000);
+ assertEquals(2000, count);
}
// ------------------------------------------------------------------------------------------------
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
similarity index 63%
copy from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
copy to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
index 3d7e6346981..612733f5f91 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.iotdb.db.queryengine.execution.operator;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
@@ -26,58 +27,44 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import org.apache.iotdb.db.queryengine.common.QueryId;
-import org.apache.iotdb.db.queryengine.common.header.DatasetHeader;
-import org.apache.iotdb.db.queryengine.common.header.DatasetHeaderFactory;
import org.apache.iotdb.db.queryengine.execution.driver.DriverContext;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.queryengine.execution.operator.process.DeviceViewOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceViewOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.queryengine.execution.operator.source.SeriesScanOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.source.ShowQueriesOperator;
-import org.apache.iotdb.db.queryengine.plan.Coordinator;
-import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
-import org.apache.iotdb.db.queryengine.plan.execution.IQueryExecution;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
-import org.apache.iotdb.db.queryengine.plan.statement.Statement;
import org.apache.iotdb.db.queryengine.plan.statement.component.OrderByKey;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import org.apache.iotdb.db.queryengine.plan.statement.component.SortItem;
import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
import org.apache.iotdb.db.storageengine.dataregion.read.reader.series.SeriesReaderTestUtil;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
-import org.apache.iotdb.db.utils.datastructure.SortKey;
import org.apache.iotdb.tsfile.exception.write.WriteProcessException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
-import com.google.common.collect.ImmutableList;
import io.airlift.units.Duration;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.Mockito;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.Objects;
-import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@@ -86,28 +73,33 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-public class MergeSortOperatorTest {
-
- private static final String MERGE_SORT_OPERATOR_TEST_SG = "root.MergeSortOperatorTest";
+/**
+ * The original data and structure of this test is same as {@link
+ * org.apache.iotdb.db.queryengine.execution.operator.MergeSortOperatorTest}
+ */
+public class TopKOperatorTest {
+ private static final String TOP_K_OPERATOR_TEST_SG = "root.TopKOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<MeasurementSchema> measurementSchemas = new ArrayList<>();
private final List<TsFileResource> seqResources = new ArrayList<>();
private final List<TsFileResource> unSeqResources = new ArrayList<>();
- private static final String DEVICE0 = MERGE_SORT_OPERATOR_TEST_SG + ".device0";
- private static final String DEVICE1 = MERGE_SORT_OPERATOR_TEST_SG + ".device1";
- private static final String DEVICE2 = MERGE_SORT_OPERATOR_TEST_SG + ".device2";
- private static final String DEVICE3 = MERGE_SORT_OPERATOR_TEST_SG + ".device3";
+ private static final String DEVICE0 = TOP_K_OPERATOR_TEST_SG + ".device0";
+ private static final String DEVICE1 = TOP_K_OPERATOR_TEST_SG + ".device1";
+ private static final String DEVICE2 = TOP_K_OPERATOR_TEST_SG + ".device2";
+ private static final String DEVICE3 = TOP_K_OPERATOR_TEST_SG + ".device3";
private int dataNodeId;
+ private final int limitValue = 100;
+
@Before
public void setUp() throws MetadataException, IOException, WriteProcessException {
dataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0);
SeriesReaderTestUtil.setUp(
- measurementSchemas, deviceIds, seqResources, unSeqResources, MERGE_SORT_OPERATOR_TEST_SG);
+ measurementSchemas, deviceIds, seqResources, unSeqResources, TOP_K_OPERATOR_TEST_SG);
}
@After
@@ -131,7 +123,7 @@ public class MergeSortOperatorTest {
// ------------------------------------------------------------------------------------------------
// order by time - 1
// ------------------------------------------------------------------------------------------------
- // MergeSortOperator
+ // TopKOperator
// ____________|_______________
// / | \
// SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
@@ -140,7 +132,7 @@ public class MergeSortOperatorTest {
// / \ / \
// SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
// ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TopKOperator topKOperatorTest(Ordering timeOrdering, Ordering deviceOrdering, int topK) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
try {
@@ -176,18 +168,18 @@ public class MergeSortOperatorTest {
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 11, new PlanNodeId("11"), MergeSortOperator.class.getSimpleName());
+ 11, new PlanNodeId("11"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
@@ -318,8 +310,8 @@ public class MergeSortOperatorTest {
Arrays.asList(4, 5),
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(10),
Arrays.asList(
singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
@@ -329,11 +321,10 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ topK);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -343,12 +334,12 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -362,7 +353,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -370,7 +361,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -378,7 +369,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -386,17 +377,17 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByTime2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -409,7 +400,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(3).isNull(i));
assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertTrue(tsBlock.getColumn(1).isNull(i));
@@ -417,7 +408,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertTrue(tsBlock.getColumn(2).isNull(i));
@@ -426,7 +417,7 @@ public class MergeSortOperatorTest {
assertTrue(tsBlock.getColumn(5).isNull(i));
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
// make sure the device column is by desc
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -434,123 +425,11 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 1500);
+ assertEquals(limitValue, count);
}
- @Test
- public void testOrderByTime3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(6, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- // make sure the device column is by desc
- assertEquals(checkDevice, 2);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 1500);
- }
-
- @Test
- public void testOrderByTime4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(6, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- // make sure the device column is by asc
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(4).isNull(i));
- assertTrue(tsBlock.getColumn(5).isNull(i));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(3).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertTrue(tsBlock.getColumn(1).isNull(i));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- assertTrue(tsBlock.getColumn(3).isNull(i));
- assertEquals(tsBlock.getColumn(4).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(5).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 1500);
- }
-
- // ------------------------------------------------------------------------------------------------
- // order by time - 2
- // ------------------------------------------------------------------------------------------------
- // [SSO]:SeriesScanOperator
- // [SDO]:SingleDeviceOperator
- // MergeSortOperator
- // ___________|___________
- // / \
- // MergeSortOperator MergeSortOperator
- // / | / \
- // [SDO] [SDO] [SDO] [SDO]
- // | | | |
- // [SSO] RowBasedTimeJoinOperator RowBasedTimeJoinOperator
- // RowBasedTimeJoinOperator
- // / \ / \ / \
- // [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
- // ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest2(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TopKOperator topKOperatorTest2(
+ Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
try {
@@ -594,26 +473,26 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(
14, new PlanNodeId("14"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 15, new PlanNodeId("15"), MergeSortOperator.class.getSimpleName());
+ 15, new PlanNodeId("15"), TopKOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 16, new PlanNodeId("16"), MergeSortOperator.class.getSimpleName());
+ 16, new PlanNodeId("16"), TopKOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 17, new PlanNodeId("17"), MergeSortOperator.class.getSimpleName());
+ 17, new PlanNodeId("17"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
MeasurementPath measurementPath6 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
MeasurementPath measurementPath7 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
@@ -787,8 +666,8 @@ public class MergeSortOperatorTest {
Arrays.asList(1, 2),
tsDataTypes);
- MergeSortOperator mergeSortOperator1 =
- new MergeSortOperator(
+ TopKOperator topKOperator1 =
+ new TopKOperator(
driverContext.getOperatorContexts().get(14),
Arrays.asList(singleDeviceViewOperator1, singleDeviceViewOperator2),
tsDataTypes,
@@ -797,12 +676,11 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator1
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- MergeSortOperator mergeSortOperator2 =
- new MergeSortOperator(
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue);
+ topKOperator1.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ TopKOperator topKOperator2 =
+ new TopKOperator(
driverContext.getOperatorContexts().get(15),
Arrays.asList(singleDeviceViewOperator3, singleDeviceViewOperator4),
tsDataTypes,
@@ -811,26 +689,24 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator2
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue);
+ topKOperator2.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator topKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(16),
- Arrays.asList(mergeSortOperator1, mergeSortOperator2),
+ Arrays.asList(topKOperator1, topKOperator2),
tsDataTypes,
MergeSortComparator.getComparator(
Arrays.asList(
new SortItem(OrderByKey.TIME, timeOrdering),
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
- Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.INT64, TSDataType.TEXT)),
+ limitValue);
+ topKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return topKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -840,12 +716,12 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime1_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -856,22 +732,22 @@ public class MergeSortOperatorTest {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by asc
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -879,18 +755,18 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByTime2_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -900,67 +776,23 @@ public class MergeSortOperatorTest {
if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by desc
- assertEquals(checkDevice, 3);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByTime3_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 0);
+ assertEquals(0, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
+ assertEquals(1, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
+ assertEquals(2, checkDevice);
checkDevice++;
} else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
assertTrue(tsBlock.getColumn(2).isNull(i));
// make sure the device column is by desc
- assertEquals(checkDevice, 3);
+ assertEquals(3, checkDevice);
checkDevice = 0;
} else {
fail();
@@ -968,57 +800,14 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
- @Test
- public void testOrderByTime4_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by asc
- assertEquals(checkDevice, 0);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 1);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 2);
- checkDevice++;
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertEquals(checkDevice, 3);
- checkDevice = 0;
- } else {
- fail();
- }
- }
- }
-
- assertEquals(count, 2000);
- }
// ------------------------------------------------------------------------------------------------
// order by device
// ------------------------------------------------------------------------------------------------
// [SSO]:SeriesScanOperator
- // MergeSortOperator
+ // TopKOperator
// ___________|___________
// / \
// DeviceViewOperator DeviceViewOperator
@@ -1028,7 +817,8 @@ public class MergeSortOperatorTest {
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest3(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TopKOperator topKOperatorTest3(
+ Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
try {
@@ -1068,22 +858,22 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(
12, new PlanNodeId("12"), DeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 13, new PlanNodeId("13"), MergeSortOperator.class.getSimpleName());
+ 13, new PlanNodeId("13"), TopKOperator.class.getSimpleName());
MeasurementPath measurementPath1 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device0.sensor0", TSDataType.INT32);
MeasurementPath measurementPath2 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor0", TSDataType.INT32);
MeasurementPath measurementPath3 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device1.sensor1", TSDataType.INT32);
MeasurementPath measurementPath4 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor0", TSDataType.INT32);
MeasurementPath measurementPath5 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device2.sensor1", TSDataType.INT32);
MeasurementPath measurementPath6 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor0", TSDataType.INT32);
MeasurementPath measurementPath7 =
- new MeasurementPath(MERGE_SORT_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
+ new MeasurementPath(TOP_K_OPERATOR_TEST_SG + ".device3.sensor1", TSDataType.INT32);
SeriesScanOperator seriesScanOperator1 =
new SeriesScanOperator(
@@ -1260,8 +1050,8 @@ public class MergeSortOperatorTest {
: Arrays.asList(timeJoinOperator3, timeJoinOperator2),
deviceColumnIndex,
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TopKOperator TopKOperator =
+ new TopKOperator(
driverContext.getOperatorContexts().get(12),
Arrays.asList(deviceViewOperator1, deviceViewOperator2),
tsDataTypes,
@@ -1270,11 +1060,10 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.DEVICE, deviceOrdering),
new SortItem(OrderByKey.TIME, timeOrdering)),
Arrays.asList(0, -1),
- Arrays.asList(TSDataType.TEXT, TSDataType.INT64)));
- mergeSortOperator
- .getOperatorContext()
- .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return mergeSortOperator;
+ Arrays.asList(TSDataType.TEXT, TSDataType.INT64)),
+ limitValue);
+ TopKOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return TopKOperator;
} catch (IllegalPathException e) {
e.printStackTrace();
fail();
@@ -1284,12 +1073,12 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByDevice1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.ASC);
+ TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1332,17 +1121,17 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
+ assertEquals(limitValue, count);
}
@Test
public void testOrderByDevice2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.DESC);
+ TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
+ TsBlock tsBlock = topKOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1385,323 +1174,6 @@ public class MergeSortOperatorTest {
}
}
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByDevice3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.ASC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
-
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by asc
- assertTrue(checkDevice < 500);
- checkDevice++;
- if (checkDevice == 500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 500 && checkDevice < 1000);
- checkDevice++;
- if (checkDevice == 1000) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1000 && checkDevice < 1500);
- checkDevice++;
- if (checkDevice == 1500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1500 && checkDevice < 2000);
- checkDevice++;
- } else {
- fail();
- }
- }
- }
- assertEquals(count, 2000);
- }
-
- @Test
- public void testOrderByDevice4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.DESC);
- long lastTime = Long.MAX_VALUE;
- int checkDevice = 0;
- int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
- if (tsBlock == null) continue;
- assertEquals(3, tsBlock.getValueColumnCount());
- count += tsBlock.getPositionCount();
- for (int i = 0; i < tsBlock.getPositionCount(); i++) {
- assertTrue(tsBlock.getTimeByIndex(i) <= lastTime);
- lastTime = tsBlock.getTimeByIndex(i);
- if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE3)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice < 500);
- checkDevice++;
- if (checkDevice == 500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE2)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 500 && checkDevice < 1000);
- checkDevice++;
- if (checkDevice == 1000) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE1)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertEquals(tsBlock.getColumn(2).getInt(i), getValue(lastTime));
- assertTrue(checkDevice >= 1000 && checkDevice < 1500);
- checkDevice++;
- if (checkDevice == 1500) {
- lastTime = Long.MAX_VALUE;
- }
- } else if (Objects.equals(tsBlock.getColumn(0).getBinary(i).toString(), DEVICE0)) {
- assertEquals(tsBlock.getColumn(1).getInt(i), getValue(lastTime));
- assertTrue(tsBlock.getColumn(2).isNull(i));
- // make sure the device column is by desc
- assertTrue(checkDevice >= 1500 && checkDevice < 2000);
- checkDevice++;
- } else {
- fail();
- }
- }
- }
- assertEquals(count, 2000);
- }
-
- // ------------------------------------------------------------------------------------------------
- // order by Time, DataNodeID
- // ------------------------------------------------------------------------------------------------
- //
- // MergeSortOperator
- // ___________|__________
- // / \
- // SortOperator SortOperator
- // | |
- // ShowQueriesOperator ShowQueriesOperator
- // ------------------------------------------------------------------------------------------------
- @Test
- public void mergeSortWithSortOperatorTest() throws Exception {
- ExecutorService instanceNotificationExecutor =
- IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
-
- try {
- // Construct operator tree
- QueryId queryId = new QueryId("stub_query");
-
- FragmentInstanceId instanceId =
- new FragmentInstanceId(new PlanFragmentId(queryId, 0), "stub-instance");
- FragmentInstanceStateMachine stateMachine =
- new FragmentInstanceStateMachine(instanceId, instanceNotificationExecutor);
- FragmentInstanceContext fragmentInstanceContext =
- createFragmentInstanceContext(instanceId, stateMachine);
- DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0);
- PlanNodeId planNodeId0 = new PlanNodeId("0");
- driverContext.addOperatorContext(0, planNodeId0, ShowQueriesOperator.class.getSimpleName());
- PlanNodeId planNodeId1 = new PlanNodeId("1");
- driverContext.addOperatorContext(1, planNodeId1, ShowQueriesOperator.class.getSimpleName());
- PlanNodeId planNodeId2 = new PlanNodeId("2");
- driverContext.addOperatorContext(2, planNodeId2, SortOperator.class.getSimpleName());
- PlanNodeId planNodeId3 = new PlanNodeId("3");
- driverContext.addOperatorContext(3, planNodeId3, SortOperator.class.getSimpleName());
- PlanNodeId planNodeId4 = new PlanNodeId("4");
- driverContext.addOperatorContext(4, planNodeId4, MergeSortOperator.class.getSimpleName());
-
- List<OperatorContext> operatorContexts = driverContext.getOperatorContexts();
- List<TSDataType> dataTypes = DatasetHeaderFactory.getShowQueriesHeader().getRespDataTypes();
- Comparator<SortKey> comparator =
- MergeSortComparator.getComparator(
- Arrays.asList(
- new SortItem(OrderByKey.TIME, Ordering.ASC),
- new SortItem(OrderByKey.DATANODEID, Ordering.DESC)),
- ImmutableList.of(-1, 1),
- ImmutableList.of(TSDataType.INT64, TSDataType.INT32));
-
- Coordinator coordinator1 = Mockito.mock(Coordinator.class);
- Mockito.when(coordinator1.getAllQueryExecutions())
- .thenReturn(
- ImmutableList.of(
- new FakeQueryExecution(3, "20221229_000000_00003_1", "sql3_node1"),
- new FakeQueryExecution(1, "20221229_000000_00001_1", "sql1_node1"),
- new FakeQueryExecution(2, "20221229_000000_00002_1", "sql2_node1")));
- Coordinator coordinator2 = Mockito.mock(Coordinator.class);
- Mockito.when(coordinator2.getAllQueryExecutions())
- .thenReturn(
- ImmutableList.of(
- new FakeQueryExecution(3, "20221229_000000_00003_2", "sql3_node2"),
- new FakeQueryExecution(2, "20221229_000000_00002_2", "sql2_node2"),
- new FakeQueryExecution(1, "20221229_000000_00001_2", "sql1_node2")));
-
- ShowQueriesOperator showQueriesOperator1 =
- new ShowQueriesOperator(operatorContexts.get(0), planNodeId0, coordinator1);
- ShowQueriesOperator showQueriesOperator2 =
- new ShowQueriesOperator(operatorContexts.get(1), planNodeId1, coordinator2);
- SortOperator sortOperator1 =
- new SortOperator(
- operatorContexts.get(2), showQueriesOperator1, dataTypes, "", comparator);
- SortOperator sortOperator2 =
- new SortOperator(
- operatorContexts.get(3), showQueriesOperator2, dataTypes, "", comparator);
- Operator root =
- new MergeSortOperator(
- operatorContexts.get(4),
- new ArrayList<Operator>() {
- {
- add(sortOperator1);
- add(sortOperator2);
- }
- },
- dataTypes,
- comparator);
- root.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
-
- int index = 0;
- // Time ASC
- long[] expectedTime = new long[] {1, 1, 2, 2, 3, 3};
- // DataNodeId DESC if Times are equal
- String[] expectedQueryId =
- new String[] {
- "20221229_000000_00001_2",
- "20221229_000000_00001_1",
- "20221229_000000_00002_2",
- "20221229_000000_00002_1",
- "20221229_000000_00003_2",
- "20221229_000000_00003_1"
- };
- int[] expectedDataNodeId = new int[] {2, 1, 2, 1, 2, 1};
- String[] expectedStatement =
- new String[] {
- "sql1_node2", "sql1_node1", "sql2_node2", "sql2_node1", "sql3_node2", "sql3_node1"
- };
- while (root.isBlocked().isDone() && root.hasNext()) {
- TsBlock result = root.next();
- if (result == null) {
- continue;
- }
-
- for (int i = 0; i < result.getPositionCount(); i++, index++) {
- assertEquals(expectedTime[index], result.getTimeColumn().getLong(i));
- assertEquals(expectedQueryId[index], result.getColumn(0).getBinary(i).toString());
- assertEquals(expectedDataNodeId[index], result.getColumn(1).getInt(i));
- assertEquals(expectedStatement[index], result.getColumn(3).getBinary(i).toString());
- }
- }
- } finally {
- instanceNotificationExecutor.shutdown();
- }
- }
-
- static class FakeQueryExecution implements IQueryExecution {
- private final long startTime;
- private final String queryId;
- private final String sql;
-
- FakeQueryExecution(long startTime, String queryId, String sql) {
- this.startTime = startTime;
- this.queryId = queryId;
- this.sql = sql;
- }
-
- @Override
- public String getQueryId() {
- return queryId;
- }
-
- @Override
- public long getStartExecutionTime() {
- return startTime;
- }
-
- @Override
- public void recordExecutionTime(long executionTime) {}
-
- @Override
- public long getTotalExecutionTime() {
- return 0;
- }
-
- @Override
- public Optional<String> getExecuteSQL() {
- return Optional.of(sql);
- }
-
- @Override
- public Statement getStatement() {
- return null;
- }
-
- @Override
- public void start() {}
-
- @Override
- public void stop(Throwable t) {}
-
- @Override
- public void stopAndCleanup() {}
-
- @Override
- public void stopAndCleanup(Throwable t) {}
-
- @Override
- public void cancel() {}
-
- @Override
- public ExecutionResult getStatus() {
- return null;
- }
-
- @Override
- public Optional<TsBlock> getBatchResult() {
- return Optional.empty();
- }
-
- @Override
- public Optional<ByteBuffer> getByteBufferBatchResult() {
- return Optional.empty();
- }
-
- @Override
- public boolean hasNextResult() {
- return false;
- }
-
- @Override
- public int getOutputValueColumnCount() {
- return 0;
- }
-
- @Override
- public DatasetHeader getDatasetHeader() {
- return null;
- }
-
- @Override
- public boolean isQuery() {
- return false;
- }
+ assertEquals(limitValue, count);
}
}