You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2023/02/22 12:39:42 UTC
[iotdb] branch master updated: [IOTDB-5546] Allow consumeAllOperator to run when only some of children is blocked
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 99ea2c9f7e [IOTDB-5546] Allow consumeAllOperator to run when only some of children is blocked
99ea2c9f7e is described below
commit 99ea2c9f7e282507e001349718d72651ab8fc28e
Author: Xiangwei Wei <34...@users.noreply.github.com>
AuthorDate: Wed Feb 22 20:39:36 2023 +0800
[IOTDB-5546] Allow consumeAllOperator to run when only some of children is blocked
---
.../db/it/aggregation/IoTDBTagAggregation2IT.java | 51 +++++++++
.../db/it/aggregation/IoTDBTagAggregationIT.java | 4 +-
.../process/AbstractConsumeAllOperator.java | 126 +++++++++++++++++++++
.../operator/process/AbstractProcessOperator.java | 24 ----
.../operator/process/AggregationOperator.java | 78 +------------
.../operator/process/DeviceMergeOperator.java | 1 +
.../operator/process/MergeSortOperator.java | 103 +++++++++--------
.../operator/process/TagAggregationOperator.java | 93 +++------------
.../process/join/HorizontallyConcatOperator.java | 68 +++--------
.../process/join/RowBasedTimeJoinOperator.java | 125 ++++++++++----------
.../operator/process/join/TimeJoinOperator.java | 4 +-
.../iotdb/db/mpp/execution/DataDriverTest.java | 8 +-
.../operator/AlignedSeriesScanOperatorTest.java | 18 +--
.../operator/HorizontallyConcatOperatorTest.java | 3 +-
.../mpp/execution/operator/LimitOperatorTest.java | 10 +-
.../execution/operator/MergeSortOperatorTest.java | 82 +++++++-------
.../mpp/execution/operator/OffsetOperatorTest.java | 26 ++---
.../mpp/execution/operator/OperatorMemoryTest.java | 37 ------
.../operator/RawDataAggregationOperatorTest.java | 47 +++++---
.../operator/SingleDeviceViewOperatorTest.java | 8 +-
.../execution/operator/TimeJoinOperatorTest.java | 3 +-
21 files changed, 453 insertions(+), 466 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregation2IT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregation2IT.java
new file mode 100644
index 0000000000..4ca6f566db
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregation2IT.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.it.aggregation;
+
+import org.apache.iotdb.it.env.EnvFactory;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import java.sql.Connection;
+import java.sql.Statement;
+
+public class IoTDBTagAggregation2IT extends IoTDBTagAggregationIT {
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ EnvFactory.getEnv().getConfig().getCommonConfig().setDegreeOfParallelism(4);
+ EnvFactory.getEnv().initClusterEnvironment();
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ for (String sql : DATASET) {
+ statement.execute(sql);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw e;
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java
index b26a5b74c3..9ca72958e5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/aggregation/IoTDBTagAggregationIT.java
@@ -45,7 +45,7 @@ import static org.junit.Assert.fail;
@RunWith(IoTDBTestRunner.class)
@Category({ClusterIT.class})
public class IoTDBTagAggregationIT {
- private static final String[] DATASET =
+ protected static final String[] DATASET =
new String[] {
"CREATE DATABASE root.sg.a;",
"CREATE DATABASE root.sg.b;",
@@ -75,7 +75,7 @@ public class IoTDBTagAggregationIT {
"insert into root.sg.a.d7(time, t) values(10, 8.7);"
};
- private static final double DELTA = 0.001D;
+ protected static final double DELTA = 0.001D;
@BeforeClass
public static void setUp() throws Exception {
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractConsumeAllOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractConsumeAllOperator.java
new file mode 100644
index 0000000000..1ae0fbf801
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractConsumeAllOperator.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.mpp.execution.operator.process;
+
+import org.apache.iotdb.db.mpp.execution.operator.AbstractOperator;
+import org.apache.iotdb.db.mpp.execution.operator.Operator;
+import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
+import org.apache.iotdb.tsfile.read.common.block.TsBlock;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static com.google.common.util.concurrent.Futures.successfulAsList;
+
+/** ConsumeAllOperator will consume all children's result every time. */
+public abstract class AbstractConsumeAllOperator extends AbstractOperator
+ implements ProcessOperator {
+ protected final List<Operator> children;
+ protected final int inputOperatorsCount;
+ /** TsBlock from child operator. Only one cache now. */
+ protected final TsBlock[] inputTsBlocks;
+
+ protected final boolean[] canCallNext;
+ protected int readyChildIndex;
+
+ protected AbstractConsumeAllOperator(OperatorContext operatorContext, List<Operator> children) {
+ this.operatorContext = operatorContext;
+ this.children = children;
+ this.inputOperatorsCount = children.size();
+ this.inputTsBlocks = new TsBlock[inputOperatorsCount];
+ this.canCallNext = new boolean[inputOperatorsCount];
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ canCallNext[i] = false;
+ }
+ }
+
+ @Override
+ public ListenableFuture<?> isBlocked() {
+ boolean hasReadyChild = false;
+ readyChildIndex = 0;
+ List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (!isEmpty(i)) {
+ continue;
+ }
+ ListenableFuture<?> blocked = children.get(i).isBlocked();
+ if (blocked.isDone()) {
+ hasReadyChild = true;
+ canCallNext[i] = true;
+ readyChildIndex = i;
+ } else {
+ listenableFutures.add(blocked);
+ }
+ }
+ return (hasReadyChild || listenableFutures.isEmpty())
+ ? NOT_BLOCKED
+ : successfulAsList(listenableFutures);
+ }
+
+ /**
+ * Try to cache one result of each child.
+ *
+ * @return true if results of all children are ready. Return false if some children is blocked or
+ * return null.
+ */
+ protected boolean prepareInput() {
+ boolean allReady = true;
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (!isEmpty(i)) {
+ continue;
+ }
+ if (canCallNext[i] && children.get(i).hasNextWithTimer()) {
+ inputTsBlocks[i] = getNextTsBlock(i);
+ canCallNext[i] = false;
+ // child operator has next but return an empty TsBlock which means that it may not
+ // finish calculation in given time slice.
+ // In such case, TimeJoinOperator can't go on calculating, so we just return null.
+ // We can also use the while loop here to continuously call the hasNext() and next()
+ // methods of the child operator until its hasNext() returns false or the next() gets
+ // the data that is not empty, but this will cause the execution time of the while loop
+ // to be uncontrollable and may exceed all allocated time slice
+ if (isEmpty(i)) {
+ allReady = false;
+ }
+ } else {
+ allReady = false;
+ }
+ }
+ return allReady;
+ }
+
+ /** If the tsBlock is null or has no more data in the tsBlock, return true; else return false. */
+ protected boolean isEmpty(int index) {
+ return inputTsBlocks[index] == null || inputTsBlocks[index].isEmpty();
+ }
+
+ @Override
+ public void close() throws Exception {
+ for (Operator child : children) {
+ child.close();
+ }
+ }
+
+ protected TsBlock getNextTsBlock(int childIndex) {
+ return children.get(childIndex).nextWithTimer();
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractProcessOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractProcessOperator.java
deleted file mode 100644
index 2e024256aa..0000000000
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AbstractProcessOperator.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.mpp.execution.operator.process;
-
-import org.apache.iotdb.db.mpp.execution.operator.AbstractOperator;
-
-public abstract class AbstractProcessOperator extends AbstractOperator implements ProcessOperator {}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
index c320839813..62b5e7a53c 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/AggregationOperator.java
@@ -27,14 +27,11 @@ import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
import org.apache.iotdb.tsfile.read.common.block.TsBlockBuilder;
-import com.google.common.util.concurrent.ListenableFuture;
-
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import static com.google.common.util.concurrent.Futures.successfulAsList;
import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.appendAggregationResult;
/**
@@ -42,14 +39,7 @@ import static org.apache.iotdb.db.mpp.execution.operator.AggregationUtil.appendA
* will output one tsBlock contain many results on aggregation time intervals. One intermediate
* tsBlock input will contain the result of many time intervals.
*/
-public class AggregationOperator implements ProcessOperator {
-
- private final OperatorContext operatorContext;
-
- private final List<Operator> children;
- private final int inputOperatorsCount;
- private final TsBlock[] inputTsBlocks;
- private final boolean[] canCallNext;
+public class AggregationOperator extends AbstractConsumeAllOperator {
private final ITimeRangeIterator timeRangeIterator;
// current interval of aggregation window [curStartTime, curEndTime)
@@ -62,7 +52,6 @@ public class AggregationOperator implements ProcessOperator {
private final long maxRetainedSize;
private final long childrenRetainedSize;
- private final long maxReturnSize;
public AggregationOperator(
OperatorContext operatorContext,
@@ -70,18 +59,9 @@ public class AggregationOperator implements ProcessOperator {
ITimeRangeIterator timeRangeIterator,
List<Operator> children,
long maxReturnSize) {
- this.operatorContext = operatorContext;
- this.children = children;
+ super(operatorContext, children);
this.aggregators = aggregators;
this.timeRangeIterator = timeRangeIterator;
-
- this.inputOperatorsCount = children.size();
- this.inputTsBlocks = new TsBlock[inputOperatorsCount];
- this.canCallNext = new boolean[inputOperatorsCount];
- for (int i = 0; i < inputOperatorsCount; i++) {
- canCallNext[i] = false;
- }
-
List<TSDataType> dataTypes = new ArrayList<>();
for (Aggregator aggregator : aggregators) {
dataTypes.addAll(Arrays.asList(aggregator.getOutputType()));
@@ -98,11 +78,6 @@ public class AggregationOperator implements ProcessOperator {
this.maxReturnSize = maxReturnSize;
}
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
@Override
public long calculateMaxPeekMemory() {
return maxReturnSize + maxRetainedSize + childrenRetainedSize;
@@ -118,23 +93,6 @@ public class AggregationOperator implements ProcessOperator {
return maxRetainedSize + childrenRetainedSize;
}
- @Override
- public ListenableFuture<?> isBlocked() {
- List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
- for (int i = 0; i < inputOperatorsCount; i++) {
- ListenableFuture<?> blocked = children.get(i).isBlocked();
- if (blocked.isDone()) {
- canCallNext[i] = true;
- } else {
- if (isEmpty(i)) {
- listenableFutures.add(blocked);
- canCallNext[i] = true;
- }
- }
- }
- return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
- }
-
@Override
public boolean hasNext() {
return curTimeRange != null || timeRangeIterator.hasNextTimeRange();
@@ -149,8 +107,7 @@ public class AggregationOperator implements ProcessOperator {
while (System.nanoTime() - start < maxRuntime
&& (curTimeRange != null || timeRangeIterator.hasNextTimeRange())
&& !resultTsBlockBuilder.isFull()) {
- boolean hasCachedData = prepareInput();
- if (!hasCachedData) {
+ if (!prepareInput()) {
break;
}
@@ -182,31 +139,6 @@ public class AggregationOperator implements ProcessOperator {
return !this.hasNextWithTimer();
}
- @Override
- public void close() throws Exception {
- for (Operator child : children) {
- child.close();
- }
- }
-
- private boolean prepareInput() {
- for (int i = 0; i < inputOperatorsCount; i++) {
- if (inputTsBlocks[i] != null) {
- continue;
- }
- if (!canCallNext[i]) {
- return false;
- }
-
- inputTsBlocks[i] = children.get(i).nextWithTimer();
- canCallNext[i] = false;
- if (inputTsBlocks[i] == null) {
- return false;
- }
- }
- return true;
- }
-
private void calculateNextAggregationResult() {
// consume current input tsBlocks
for (Aggregator aggregator : aggregators) {
@@ -229,8 +161,4 @@ public class AggregationOperator implements ProcessOperator {
appendAggregationResult(
resultTsBlockBuilder, aggregators, timeRangeIterator.currentOutputTime());
}
-
- private boolean isEmpty(int index) {
- return inputTsBlocks[index] == null || inputTsBlocks[index].isEmpty();
- }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceMergeOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceMergeOperator.java
index c08895ceaa..888db23336 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceMergeOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/DeviceMergeOperator.java
@@ -49,6 +49,7 @@ import static com.google.common.util.concurrent.Futures.successfulAsList;
* <p>The form of tsBlocks from input operators should be the same strictly, which is transferred by
* DeviceViewOperator.
*/
+@Deprecated
public class DeviceMergeOperator implements ProcessOperator {
private final OperatorContext operatorContext;
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/MergeSortOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/MergeSortOperator.java
index db11810fa2..2235bc7c91 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/MergeSortOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/MergeSortOperator.java
@@ -37,14 +37,10 @@ import java.util.List;
import static com.google.common.util.concurrent.Futures.successfulAsList;
-public class MergeSortOperator implements ProcessOperator {
+public class MergeSortOperator extends AbstractConsumeAllOperator {
- private final OperatorContext operatorContext;
- private final List<Operator> inputOperators;
private final List<TSDataType> dataTypes;
private final TsBlockBuilder tsBlockBuilder;
- private final int inputOperatorsCount;
- private final TsBlock[] inputTsBlocks;
private final boolean[] noMoreTsBlocks;
private final MergeSortHeap mergeSortHeap;
private final Comparator<MergeSortKey> comparator;
@@ -56,53 +52,40 @@ public class MergeSortOperator implements ProcessOperator {
List<Operator> inputOperators,
List<TSDataType> dataTypes,
Comparator<MergeSortKey> comparator) {
- this.operatorContext = operatorContext;
- this.inputOperators = inputOperators;
+ super(operatorContext, inputOperators);
this.dataTypes = dataTypes;
- this.inputOperatorsCount = inputOperators.size();
this.mergeSortHeap = new MergeSortHeap(inputOperatorsCount, comparator);
this.comparator = comparator;
- this.inputTsBlocks = new TsBlock[inputOperatorsCount];
this.noMoreTsBlocks = new boolean[inputOperatorsCount];
this.tsBlockBuilder = new TsBlockBuilder(dataTypes);
}
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
@Override
public ListenableFuture<?> isBlocked() {
+ boolean hasReadyChild = false;
List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!noMoreTsBlocks[i] && isTsBlockEmpty(i)) {
- ListenableFuture<?> blocked = inputOperators.get(i).isBlocked();
- if (!blocked.isDone()) {
- listenableFutures.add(blocked);
- }
+ if (noMoreTsBlocks[i] || !isEmpty(i)) {
+ continue;
+ }
+ ListenableFuture<?> blocked = children.get(i).isBlocked();
+ if (blocked.isDone()) {
+ hasReadyChild = true;
+ canCallNext[i] = true;
+ } else {
+ listenableFutures.add(blocked);
}
}
- return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
- }
-
- /** If the tsBlock is null or has no more data in the tsBlock, return true; else return false; */
- private boolean isTsBlockEmpty(int tsBlockIndex) {
- return inputTsBlocks[tsBlockIndex] == null
- || inputTsBlocks[tsBlockIndex].getPositionCount() == 0;
+ return (hasReadyChild || listenableFutures.isEmpty())
+ ? NOT_BLOCKED
+ : successfulAsList(listenableFutures);
}
@Override
public TsBlock next() {
// 1. fill consumed up TsBlock
- for (int i = 0; i < inputOperatorsCount; i++) {
- if (!noMoreTsBlocks[i] && isTsBlockEmpty(i) && inputOperators.get(i).hasNextWithTimer()) {
- inputTsBlocks[i] = inputOperators.get(i).nextWithTimer();
- if (inputTsBlocks[i] == null || inputTsBlocks[i].isEmpty()) {
- return null;
- }
- mergeSortHeap.push(new MergeSortKey(inputTsBlocks[i], 0, i));
- }
+ if (!prepareInput()) {
+ return null;
}
// 2. check if we can directly return the original TsBlock instead of merging way
@@ -157,10 +140,10 @@ public class MergeSortOperator implements ProcessOperator {
return false;
}
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!isTsBlockEmpty(i)) {
+ if (!isEmpty(i)) {
return true;
} else if (!noMoreTsBlocks[i]) {
- if (inputOperators.get(i).hasNextWithTimer()) {
+ if (!canCallNext[i] || children.get(i).hasNextWithTimer()) {
return true;
} else {
noMoreTsBlocks[i] = true;
@@ -171,13 +154,6 @@ public class MergeSortOperator implements ProcessOperator {
return false;
}
- @Override
- public void close() throws Exception {
- for (Operator operator : inputOperators) {
- operator.close();
- }
- }
-
@Override
public boolean isFinished() {
if (finished) {
@@ -186,7 +162,7 @@ public class MergeSortOperator implements ProcessOperator {
finished = true;
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!noMoreTsBlocks[i] || !isTsBlockEmpty(i)) {
+ if (!noMoreTsBlocks[i] || !isEmpty(i)) {
finished = false;
break;
}
@@ -199,11 +175,11 @@ public class MergeSortOperator implements ProcessOperator {
// MergeToolKit will cache startKey and endKey
long maxPeekMemory = TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
// inputTsBlocks will cache all the tsBlocks returned by inputOperators
- for (Operator operator : inputOperators) {
+ for (Operator operator : children) {
maxPeekMemory += operator.calculateMaxReturnSize();
maxPeekMemory += operator.calculateRetainedSizeAfterCallingNext();
}
- for (Operator operator : inputOperators) {
+ for (Operator operator : children) {
maxPeekMemory = Math.max(maxPeekMemory, operator.calculateMaxPeekMemory());
}
return Math.max(maxPeekMemory, calculateMaxReturnSize());
@@ -217,11 +193,44 @@ public class MergeSortOperator implements ProcessOperator {
@Override
public long calculateRetainedSizeAfterCallingNext() {
long currentRetainedSize = 0, minChildReturnSize = Long.MAX_VALUE;
- for (Operator child : inputOperators) {
+ for (Operator child : children) {
long maxReturnSize = child.calculateMaxReturnSize();
minChildReturnSize = Math.min(minChildReturnSize, maxReturnSize);
currentRetainedSize += (maxReturnSize + child.calculateRetainedSizeAfterCallingNext());
}
return currentRetainedSize - minChildReturnSize;
}
+
+ /**
+ * Try to cache one result of each child.
+ *
+ * @return true if results of all children are ready or have no more TsBlocks. Return false if
+ * some children is blocked or return null.
+ */
+ @Override
+ protected boolean prepareInput() {
+ boolean allReady = true;
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (noMoreTsBlocks[i] || !isEmpty(i)) {
+ 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;
+ }
+ } else {
+ allReady = false;
+ }
+ }
+ return allReady;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java
index 1e2d5850ce..9cc3e37a63 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/TagAggregationOperator.java
@@ -29,7 +29,6 @@ import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.iotdb.tsfile.utils.Binary;
-import com.google.common.util.concurrent.ListenableFuture;
import org.apache.commons.lang3.Validate;
import java.util.ArrayList;
@@ -37,25 +36,16 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
-import static com.google.common.util.concurrent.Futures.successfulAsList;
+public class TagAggregationOperator extends AbstractConsumeAllOperator {
-public class TagAggregationOperator implements ProcessOperator {
-
- private final OperatorContext operatorContext;
private final List<List<String>> groups;
private final List<List<Aggregator>> groupedAggregators;
- private final List<Operator> children;
- private final TsBlock[] inputTsBlocks;
-
- // Indicate whether a child operator's next() can be called
- private final boolean[] canCallNext;
// These fields record the to be consumed index of each tsBlock.
private final int[] consumedIndices;
private final TsBlockBuilder tsBlockBuilder;
private final long maxRetainedSize;
private final long childrenRetainedSize;
- private final long maxReturnSize;
public TagAggregationOperator(
OperatorContext operatorContext,
@@ -63,10 +53,9 @@ public class TagAggregationOperator implements ProcessOperator {
List<List<Aggregator>> groupedAggregators,
List<Operator> children,
long maxReturnSize) {
- this.operatorContext = Validate.notNull(operatorContext);
+ super(operatorContext, children);
this.groups = Validate.notNull(groups);
this.groupedAggregators = Validate.notNull(groupedAggregators);
- this.children = Validate.notNull(children);
List<TSDataType> actualOutputColumnTypes = new ArrayList<>();
for (String ignored : groups.get(0)) {
actualOutputColumnTypes.add(TSDataType.TEXT);
@@ -83,9 +72,6 @@ public class TagAggregationOperator implements ProcessOperator {
}
}
this.tsBlockBuilder = new TsBlockBuilder(actualOutputColumnTypes);
- // Initialize input tsblocks for each aggregator group.
- this.inputTsBlocks = new TsBlock[children.size()];
- this.canCallNext = new boolean[children.size()];
Arrays.fill(canCallNext, false);
this.consumedIndices = new int[children.size()];
this.maxRetainedSize = children.stream().mapToLong(Operator::calculateMaxReturnSize).sum();
@@ -94,19 +80,15 @@ public class TagAggregationOperator implements ProcessOperator {
this.maxReturnSize = maxReturnSize;
}
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
@Override
public TsBlock next() {
long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
long start = System.nanoTime();
-
- boolean successful = true;
- while (System.nanoTime() - start < maxRuntime && !tsBlockBuilder.isFull() && successful) {
- successful = processOneRow();
+ while (System.nanoTime() - start < maxRuntime && !tsBlockBuilder.isFull()) {
+ if (!prepareInput()) {
+ break;
+ }
+ processOneRow();
}
TsBlock tsBlock = null;
if (tsBlockBuilder.getPositionCount() > 0) {
@@ -116,27 +98,7 @@ public class TagAggregationOperator implements ProcessOperator {
return tsBlock;
}
- private boolean processOneRow() {
- for (int i = 0; i < children.size(); i++) {
- if (!dataUnavailable(i)) {
- continue;
- }
-
- if (!canCallNext[i] || !children.get(i).hasNextWithTimer()) {
- return false;
- }
-
- // If the data is unavailable first, try to find next tsblock of the child.
- inputTsBlocks[i] = children.get(i).nextWithTimer();
- consumedIndices[i] = 0;
- canCallNext[i] = false;
-
- // If it's still unavailable, then blocked by children i.
- if (dataUnavailable(i)) {
- return false;
- }
- }
-
+ private void processOneRow() {
TsBlock[] rowBlocks = new TsBlock[children.size()];
for (int i = 0; i < children.size(); i++) {
rowBlocks[i] = inputTsBlocks[i].getRegion(consumedIndices[i], 1);
@@ -180,17 +142,11 @@ public class TagAggregationOperator implements ProcessOperator {
for (int i = 0; i < children.size(); i++) {
consumedIndices[i]++;
}
- return true;
}
@Override
public boolean hasNext() {
- for (int i = 0; i < children.size(); i++) {
- if (dataUnavailable(i) && !children.get(i).hasNextWithTimer()) {
- return false;
- }
- }
- return true;
+ return !isEmpty(readyChildIndex) || children.get(readyChildIndex).hasNextWithTimer();
}
@Override
@@ -198,23 +154,6 @@ public class TagAggregationOperator implements ProcessOperator {
return !this.hasNextWithTimer();
}
- @Override
- public ListenableFuture<?> isBlocked() {
- List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
- for (int i = 0, size = children.size(); i < size; i++) {
- ListenableFuture<?> blocked = children.get(i).isBlocked();
- if (blocked.isDone()) {
- canCallNext[i] = true;
- } else {
- if (dataUnavailable(i)) {
- listenableFutures.add(blocked);
- canCallNext[i] = true;
- }
- }
- }
- return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
- }
-
@Override
public long calculateMaxPeekMemory() {
return maxReturnSize + maxRetainedSize + childrenRetainedSize;
@@ -231,14 +170,14 @@ public class TagAggregationOperator implements ProcessOperator {
}
@Override
- public void close() throws Exception {
- for (Operator child : children) {
- child.close();
- }
- }
-
- private boolean dataUnavailable(int index) {
+ protected boolean isEmpty(int index) {
return inputTsBlocks[index] == null
|| consumedIndices[index] == inputTsBlocks[index].getPositionCount();
}
+
+ @Override
+ protected TsBlock getNextTsBlock(int childIndex) {
+ consumedIndices[childIndex] = 0;
+ return children.get(childIndex).nextWithTimer();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/HorizontallyConcatOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/HorizontallyConcatOperator.java
index 5cd9486e3e..dadd3760af 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/HorizontallyConcatOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/HorizontallyConcatOperator.java
@@ -20,7 +20,7 @@ package org.apache.iotdb.db.mpp.execution.operator.process.join;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.ProcessOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.AbstractConsumeAllOperator;
import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.block.TsBlock;
@@ -30,13 +30,9 @@ import org.apache.iotdb.tsfile.read.common.block.column.ColumnBuilder;
import org.apache.iotdb.tsfile.read.common.block.column.TimeColumn;
import org.apache.iotdb.tsfile.read.common.block.column.TimeColumnBuilder;
-import com.google.common.util.concurrent.ListenableFuture;
-
-import java.util.ArrayList;
import java.util.List;
import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.util.concurrent.Futures.successfulAsList;
/**
* This operator is used to horizontally concatenate TsBlocks with the same timestamp column.
@@ -45,16 +41,7 @@ import static com.google.common.util.concurrent.Futures.successfulAsList;
*
* <p>HorizontallyConcat(A,B) is: [1, 1.0, true; 2, 2.0, false]
*/
-public class HorizontallyConcatOperator implements ProcessOperator {
-
- private final OperatorContext operatorContext;
-
- private final List<Operator> children;
-
- private final int inputOperatorsCount;
-
- /** TsBlock from child operator. Only one cache now. */
- private final TsBlock[] inputTsBlocks;
+public class HorizontallyConcatOperator extends AbstractConsumeAllOperator {
/** start index for each input TsBlocks and size of it is equal to inputTsBlocks */
private final int[] inputIndex;
@@ -67,51 +54,23 @@ public class HorizontallyConcatOperator implements ProcessOperator {
public HorizontallyConcatOperator(
OperatorContext operatorContext, List<Operator> children, List<TSDataType> dataTypes) {
+ super(operatorContext, children);
checkArgument(
- children != null && !children.isEmpty(),
- "child size of VerticallyConcatOperator should be larger than 0");
- this.operatorContext = operatorContext;
- this.children = children;
- this.inputOperatorsCount = children.size();
- this.inputTsBlocks = new TsBlock[this.inputOperatorsCount];
+ !children.isEmpty(), "child size of VerticallyConcatOperator should be larger than 0");
this.inputIndex = new int[this.inputOperatorsCount];
this.outputColumnCount = dataTypes.size();
this.tsBlockBuilder = new TsBlockBuilder(dataTypes);
}
- @Override
- public OperatorContext getOperatorContext() {
- return operatorContext;
- }
-
- @Override
- public ListenableFuture<?> isBlocked() {
- List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
- for (int i = 0; i < inputOperatorsCount; i++) {
- if (empty(i)) {
- ListenableFuture<?> blocked = children.get(i).isBlocked();
- if (!blocked.isDone()) {
- listenableFutures.add(blocked);
- }
- }
- }
- return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
- }
-
@Override
public TsBlock next() {
+ if (!prepareInput()) {
+ return null;
+ }
tsBlockBuilder.reset();
// indicates how many rows can be built in this calculate
int maxRowCanBuild = Integer.MAX_VALUE;
for (int i = 0; i < inputOperatorsCount; i++) {
- if (empty(i)) {
- inputIndex[i] = 0;
- inputTsBlocks[i] = children.get(i).nextWithTimer();
- if (empty(i)) {
- // child operator has not prepared TsBlock well
- return null;
- }
- }
maxRowCanBuild =
Math.min(maxRowCanBuild, inputTsBlocks[i].getPositionCount() - inputIndex[i]);
}
@@ -151,7 +110,7 @@ public class HorizontallyConcatOperator implements ProcessOperator {
if (finished) {
return false;
}
- return !empty(0) || children.get(0).hasNextWithTimer();
+ return !isEmpty(readyChildIndex) || children.get(readyChildIndex).hasNextWithTimer();
}
@Override
@@ -166,7 +125,7 @@ public class HorizontallyConcatOperator implements ProcessOperator {
if (finished) {
return true;
}
- return finished = empty(0) && !children.get(0).hasNextWithTimer();
+ return finished = isEmpty(readyChildIndex) && !children.get(readyChildIndex).hasNextWithTimer();
}
@Override
@@ -207,8 +166,15 @@ public class HorizontallyConcatOperator implements ProcessOperator {
* If the tsBlock of tsBlockIndex is null or has no more data in the tsBlock, return true; else
* return false;
*/
- private boolean empty(int tsBlockIndex) {
+ @Override
+ protected boolean isEmpty(int tsBlockIndex) {
return inputTsBlocks[tsBlockIndex] == null
|| inputTsBlocks[tsBlockIndex].getPositionCount() == inputIndex[tsBlockIndex];
}
+
+ @Override
+ protected TsBlock getNextTsBlock(int childIndex) {
+ inputIndex[childIndex] = 0;
+ return children.get(childIndex).nextWithTimer();
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/RowBasedTimeJoinOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/RowBasedTimeJoinOperator.java
index 79f9905ba8..aca522cc57 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/RowBasedTimeJoinOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/RowBasedTimeJoinOperator.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.mpp.execution.operator.process.join;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.AbstractProcessOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.AbstractConsumeAllOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.ColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.TimeComparator;
import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
@@ -40,14 +40,7 @@ import java.util.List;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.util.concurrent.Futures.successfulAsList;
-public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
-
- private final List<Operator> children;
-
- private final int inputOperatorsCount;
-
- /** TsBlock from child operator. Only one cache now. */
- private final TsBlock[] inputTsBlocks;
+public class RowBasedTimeJoinOperator extends AbstractConsumeAllOperator {
/** start index for each input TsBlocks and size of it is equal to inputTsBlocks */
private final int[] inputIndex;
@@ -87,13 +80,8 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
List<TSDataType> dataTypes,
List<ColumnMerger> mergers,
TimeComparator comparator) {
- checkArgument(
- children != null && !children.isEmpty(),
- "child size of TimeJoinOperator should be larger than 0");
- this.operatorContext = operatorContext;
- this.children = children;
- this.inputOperatorsCount = children.size();
- this.inputTsBlocks = new TsBlock[this.inputOperatorsCount];
+ super(operatorContext, children);
+ checkArgument(!children.isEmpty(), "child size of TimeJoinOperator should be larger than 0");
this.inputIndex = new int[this.inputOperatorsCount];
this.shadowInputIndex = new int[this.inputOperatorsCount];
this.noMoreTsBlocks = new boolean[this.inputOperatorsCount];
@@ -112,16 +100,23 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
@Override
public ListenableFuture<?> isBlocked() {
+ boolean hasReadyChild = false;
List<ListenableFuture<?>> listenableFutures = new ArrayList<>();
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!noMoreTsBlocks[i] && empty(i)) {
- ListenableFuture<?> blocked = children.get(i).isBlocked();
- if (!blocked.isDone()) {
- listenableFutures.add(blocked);
- }
+ if (noMoreTsBlocks[i] || !isEmpty(i)) {
+ continue;
+ }
+ ListenableFuture<?> blocked = children.get(i).isBlocked();
+ if (blocked.isDone()) {
+ hasReadyChild = true;
+ canCallNext[i] = true;
+ } else {
+ listenableFutures.add(blocked);
}
}
- return listenableFutures.isEmpty() ? NOT_BLOCKED : successfulAsList(listenableFutures);
+ return (hasReadyChild || listenableFutures.isEmpty())
+ ? NOT_BLOCKED
+ : successfulAsList(listenableFutures);
}
@Override
@@ -130,6 +125,10 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
return getResultFromRetainedTsBlock();
}
tsBlockBuilder.reset();
+ if (!prepareInput()) {
+ return null;
+ }
+
// end time for returned TsBlock this time, it's the min/max end time among all the children
// TsBlocks order by asc/desc
long currentEndTime = 0;
@@ -138,29 +137,8 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
// get TsBlock for each input, put their time stamp into TimeSelector and then use the min Time
// among all the input TsBlock as the current output TsBlock's endTime.
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!noMoreTsBlocks[i] && empty(i)) {
- if (children.get(i).hasNextWithTimer()) {
- inputIndex[i] = 0;
- inputTsBlocks[i] = children.get(i).nextWithTimer();
- if (!empty(i)) {
- updateTimeSelector(i);
- } else {
- // child operator has next but return an empty TsBlock which means that it may not
- // finish calculation in given time slice.
- // In such case, TimeJoinOperator can't go on calculating, so we just return null.
- // We can also use the while loop here to continuously call the hasNext() and next()
- // methods of the child operator until its hasNext() returns false or the next() gets
- // the data that is not empty, but this will cause the execution time of the while loop
- // to be uncontrollable and may exceed all allocated time slice
- return null;
- }
- } else { // no more tsBlock
- noMoreTsBlocks[i] = true;
- inputTsBlocks[i] = null;
- }
- }
- // update the currentEndTime if the TsBlock is not empty
- if (!empty(i)) {
+ if (!noMoreTsBlocks[i]) {
+ // update the currentEndTime if the TsBlock is not empty
currentEndTime =
init
? comparator.getCurrentEndTime(currentEndTime, inputTsBlocks[i].getEndTime())
@@ -193,7 +171,7 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
for (int i = 0; i < inputOperatorsCount; i++) {
if (inputIndex[i] != shadowInputIndex[i]) {
inputIndex[i] = shadowInputIndex[i];
- if (!empty(i)) {
+ if (!isEmpty(i)) {
updateTimeSelector(i);
}
}
@@ -214,10 +192,10 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
return true;
}
for (int i = 0; i < inputOperatorsCount; i++) {
- if (!empty(i)) {
+ if (!isEmpty(i)) {
return true;
} else if (!noMoreTsBlocks[i]) {
- if (children.get(i).hasNextWithTimer()) {
+ if (!canCallNext[i] || children.get(i).hasNextWithTimer()) {
return true;
} else {
noMoreTsBlocks[i] = true;
@@ -228,13 +206,6 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
return false;
}
- @Override
- public void close() throws Exception {
- for (Operator child : children) {
- child.close();
- }
- }
-
@Override
public boolean isFinished() {
if (finished) {
@@ -247,7 +218,7 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
finished = true;
for (int i = 0; i < inputOperatorsCount; i++) {
// has more tsBlock output from children[i] or has cached tsBlock in inputTsBlocks[i]
- if (!noMoreTsBlocks[i] || !empty(i)) {
+ if (!noMoreTsBlocks[i] || !isEmpty(i)) {
finished = false;
break;
}
@@ -291,16 +262,56 @@ public class RowBasedTimeJoinOperator extends AbstractProcessOperator {
timeSelector.add(inputTsBlocks[index].getTimeByIndex(inputIndex[index]));
}
+ /**
+ * Try to cache one result of each child.
+ *
+ * @return true if results of all children are ready or have no more TsBlocks. Return false if
+ * some children is blocked or return null.
+ */
+ @Override
+ protected boolean prepareInput() {
+ boolean allReady = true;
+ for (int i = 0; i < inputOperatorsCount; i++) {
+ if (noMoreTsBlocks[i] || !isEmpty(i)) {
+ continue;
+ }
+ if (canCallNext[i]) {
+ if (children.get(i).hasNextWithTimer()) {
+ inputTsBlocks[i] = getNextTsBlock(i);
+ canCallNext[i] = false;
+ if (isEmpty(i)) {
+ allReady = false;
+ } else {
+ updateTimeSelector(i);
+ }
+ } else {
+ noMoreTsBlocks[i] = true;
+ inputTsBlocks[i] = null;
+ }
+ } else {
+ allReady = false;
+ }
+ }
+ return allReady;
+ }
+
@TestOnly
public List<Operator> getChildren() {
return children;
}
+ @Override
+ protected TsBlock getNextTsBlock(int childIndex) {
+ inputIndex[childIndex] = 0;
+ return children.get(childIndex).nextWithTimer();
+ }
+
/**
* If the tsBlock of columnIndex is null or has no more data in the tsBlock, return true; else
* return false;
*/
- private boolean empty(int columnIndex) {
+ @Override
+ protected boolean isEmpty(int columnIndex) {
return inputTsBlocks[columnIndex] == null
|| inputTsBlocks[columnIndex].getPositionCount() == inputIndex[columnIndex];
}
diff --git a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/TimeJoinOperator.java b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/TimeJoinOperator.java
index 74bbada77c..f945b700e4 100644
--- a/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/TimeJoinOperator.java
+++ b/server/src/main/java/org/apache/iotdb/db/mpp/execution/operator/process/join/TimeJoinOperator.java
@@ -18,9 +18,9 @@
*/
package org.apache.iotdb.db.mpp.execution.operator.process.join;
+import org.apache.iotdb.db.mpp.execution.operator.AbstractOperator;
import org.apache.iotdb.db.mpp.execution.operator.Operator;
import org.apache.iotdb.db.mpp.execution.operator.OperatorContext;
-import org.apache.iotdb.db.mpp.execution.operator.process.AbstractProcessOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.ColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.TimeComparator;
import org.apache.iotdb.db.mpp.plan.statement.component.Ordering;
@@ -39,7 +39,7 @@ import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.util.concurrent.Futures.successfulAsList;
@Deprecated
-public class TimeJoinOperator extends AbstractProcessOperator {
+public class TimeJoinOperator extends AbstractOperator {
private final List<Operator> children;
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
index 157f910a0a..de1846f338 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/DataDriverTest.java
@@ -37,7 +37,7 @@ import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceState;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.mpp.execution.operator.process.LimitOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
@@ -120,7 +120,7 @@ public class DataDriverTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(4, new PlanNodeId("4"), LimitOperator.class.getSimpleName());
SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();
@@ -154,8 +154,8 @@ public class DataDriverTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesScanOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesScanOperatorTest.java
index 81846a3807..894ef7c8c3 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesScanOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/AlignedSeriesScanOperatorTest.java
@@ -31,7 +31,7 @@ import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.DescTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
@@ -213,7 +213,7 @@ public class AlignedSeriesScanOperatorTest {
PlanNodeId planNodeId8 = new PlanNodeId("8");
driverContext.addOperatorContext(8, planNodeId8, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 9, new PlanNodeId("9"), TimeJoinOperator.class.getSimpleName());
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
AlignedSeriesScanOperator seriesScanOperator1 =
new AlignedSeriesScanOperator(
driverContext.getOperatorContexts().get(0),
@@ -343,8 +343,8 @@ public class AlignedSeriesScanOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
Arrays.asList(
seriesScanOperator1,
@@ -396,7 +396,7 @@ public class AlignedSeriesScanOperatorTest {
new SingleColumnMerger(new InputLocation(7, 0), new AscTimeComparator())),
new AscTimeComparator());
int count = 0;
- while (timeJoinOperator.hasNext()) {
+ while (timeJoinOperator.isBlocked().isDone() && timeJoinOperator.hasNext()) {
TsBlock tsBlock = timeJoinOperator.next();
assertEquals(18, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof BooleanColumn);
@@ -502,7 +502,7 @@ public class AlignedSeriesScanOperatorTest {
PlanNodeId planNodeId8 = new PlanNodeId("8");
driverContext.addOperatorContext(8, planNodeId8, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 9, new PlanNodeId("9"), TimeJoinOperator.class.getSimpleName());
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
AlignedSeriesScanOperator seriesScanOperator1 =
new AlignedSeriesScanOperator(
driverContext.getOperatorContexts().get(0),
@@ -631,8 +631,8 @@ public class AlignedSeriesScanOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
Arrays.asList(
seriesScanOperator1,
@@ -685,7 +685,7 @@ public class AlignedSeriesScanOperatorTest {
new DescTimeComparator());
int count = 499;
- while (timeJoinOperator.hasNext()) {
+ while (timeJoinOperator.isBlocked().isDone() && timeJoinOperator.hasNext()) {
TsBlock tsBlock = timeJoinOperator.next();
assertEquals(18, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof BooleanColumn);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/HorizontallyConcatOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/HorizontallyConcatOperatorTest.java
index f31027811b..1430fc4a11 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/HorizontallyConcatOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/HorizontallyConcatOperatorTest.java
@@ -184,7 +184,8 @@ public class HorizontallyConcatOperatorTest {
TSDataType.INT32));
int count = 0;
- while (horizontallyConcatOperator.hasNext()) {
+ while (horizontallyConcatOperator.isBlocked().isDone()
+ && horizontallyConcatOperator.hasNext()) {
TsBlock tsBlock = horizontallyConcatOperator.next();
assertEquals(6, tsBlock.getValueColumnCount());
for (int i = 0; i < tsBlock.getPositionCount(); i++, count++) {
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LimitOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LimitOperatorTest.java
index ea0eecd5f1..79a4d56445 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LimitOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/LimitOperatorTest.java
@@ -31,7 +31,7 @@ import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.mpp.execution.operator.process.LimitOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
@@ -108,7 +108,7 @@ public class LimitOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(4, new PlanNodeId("4"), LimitOperator.class.getSimpleName());
SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();
@@ -139,8 +139,8 @@ public class LimitOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
@@ -153,7 +153,7 @@ public class LimitOperatorTest {
LimitOperator limitOperator =
new LimitOperator(driverContext.getOperatorContexts().get(3), 250, timeJoinOperator);
int count = 0;
- while (limitOperator.hasNext()) {
+ while (limitOperator.isBlocked().isDone() && limitOperator.hasNext()) {
TsBlock tsBlock = limitOperator.next();
assertEquals(2, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/MergeSortOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/MergeSortOperatorTest.java
index 4335e98807..3cc0f1ec56 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/MergeSortOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/MergeSortOperatorTest.java
@@ -38,7 +38,7 @@ import org.apache.iotdb.db.mpp.execution.operator.process.DeviceViewOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.SingleDeviceViewOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.SortOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.DescTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.MergeSortComparator;
@@ -168,11 +168,11 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(
6, new PlanNodeId("6"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 7, new PlanNodeId("7"), TimeJoinOperator.class.getSimpleName());
+ 7, new PlanNodeId("7"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
8, new PlanNodeId("8"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 9, new PlanNodeId("9"), TimeJoinOperator.class.getSimpleName());
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
@@ -266,8 +266,8 @@ public class MergeSortOperatorTest {
Collections.singletonList(1),
tsDataTypes);
- TimeJoinOperator timeJoinOperator1 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator1 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(6),
Arrays.asList(seriesScanOperator2, seriesScanOperator3),
timeOrdering,
@@ -292,8 +292,8 @@ public class MergeSortOperatorTest {
Arrays.asList(2, 3),
tsDataTypes);
- TimeJoinOperator timeJoinOperator2 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator2 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
Arrays.asList(seriesScanOperator4, seriesScanOperator5),
timeOrdering,
@@ -342,7 +342,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -388,7 +388,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -434,7 +434,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -480,7 +480,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -532,7 +532,8 @@ public class MergeSortOperatorTest {
// / | / \
// [SDO] [SDO] [SDO] [SDO]
// | | | |
- // [SSO] TimeJoinOperator TimeJoinOperator TimeJoinOperator
+ // [SSO] RowBasedTimeJoinOperator RowBasedTimeJoinOperator
+ // RowBasedTimeJoinOperator
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------------------
@@ -566,11 +567,11 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(7, planNodeId7, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 8, new PlanNodeId("8"), TimeJoinOperator.class.getSimpleName());
+ 8, new PlanNodeId("8"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 9, new PlanNodeId("9"), TimeJoinOperator.class.getSimpleName());
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 10, new PlanNodeId("10"), TimeJoinOperator.class.getSimpleName());
+ 10, new PlanNodeId("10"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
11, new PlanNodeId("11"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
@@ -688,8 +689,8 @@ public class MergeSortOperatorTest {
List<TSDataType> tsDataTypes =
new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
- TimeJoinOperator timeJoinOperator1 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator1 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(7),
Arrays.asList(seriesScanOperator2, seriesScanOperator3),
timeOrdering,
@@ -707,8 +708,8 @@ public class MergeSortOperatorTest {
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- TimeJoinOperator timeJoinOperator2 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator2 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
Arrays.asList(seriesScanOperator4, seriesScanOperator5),
timeOrdering,
@@ -726,8 +727,8 @@ public class MergeSortOperatorTest {
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- TimeJoinOperator timeJoinOperator3 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator3 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(9),
Arrays.asList(seriesScanOperator6, seriesScanOperator7),
timeOrdering,
@@ -819,7 +820,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -861,7 +862,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -903,7 +904,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -945,7 +946,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -989,7 +990,8 @@ public class MergeSortOperatorTest {
// / \
// DeviceViewOperator DeviceViewOperator
// / | / \
- // [SSO] TimeJoinOperator TimeJoinOperator TimeJoinOperator
+ // [SSO] RowBasedTimeJoinOperator RowBasedTimeJoinOperator
+ // RowBasedTimeJoinOperator
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------------------
@@ -1023,11 +1025,11 @@ public class MergeSortOperatorTest {
driverContext.addOperatorContext(7, planNodeId7, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 8, new PlanNodeId("8"), TimeJoinOperator.class.getSimpleName());
+ 8, new PlanNodeId("8"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 9, new PlanNodeId("9"), TimeJoinOperator.class.getSimpleName());
+ 9, new PlanNodeId("9"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 10, new PlanNodeId("10"), TimeJoinOperator.class.getSimpleName());
+ 10, new PlanNodeId("10"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
11, new PlanNodeId("11"), DeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
@@ -1137,8 +1139,8 @@ public class MergeSortOperatorTest {
List<TSDataType> tsDataTypes =
new LinkedList<>(Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32));
- TimeJoinOperator timeJoinOperator1 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator1 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(7),
Arrays.asList(seriesScanOperator2, seriesScanOperator3),
timeOrdering,
@@ -1156,8 +1158,8 @@ public class MergeSortOperatorTest {
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- TimeJoinOperator timeJoinOperator2 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator2 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(8),
Arrays.asList(seriesScanOperator4, seriesScanOperator5),
timeOrdering,
@@ -1175,8 +1177,8 @@ public class MergeSortOperatorTest {
: new DescTimeComparator())),
timeOrdering == Ordering.ASC ? new AscTimeComparator() : new DescTimeComparator());
- TimeJoinOperator timeJoinOperator3 =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator3 =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(9),
Arrays.asList(seriesScanOperator6, seriesScanOperator7),
timeOrdering,
@@ -1248,7 +1250,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
@@ -1300,7 +1302,7 @@ public class MergeSortOperatorTest {
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
@@ -1352,7 +1354,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
@@ -1404,7 +1406,7 @@ public class MergeSortOperatorTest {
long lastTime = Long.MAX_VALUE;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.hasNext()) {
+ while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
TsBlock tsBlock = mergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
@@ -1546,7 +1548,7 @@ public class MergeSortOperatorTest {
new String[] {
"sql1_node2", "sql1_node1", "sql2_node2", "sql2_node1", "sql3_node2", "sql3_node1"
};
- while (root.hasNext()) {
+ while (root.isBlocked().isDone() && root.hasNext()) {
TsBlock result = root.next();
if (result == null) {
continue;
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OffsetOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OffsetOperatorTest.java
index db28de20e6..16747a70ea 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OffsetOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OffsetOperatorTest.java
@@ -32,7 +32,7 @@ import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.mpp.execution.operator.process.LimitOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.OffsetOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
@@ -109,7 +109,7 @@ public class OffsetOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
4, new PlanNodeId("4"), OffsetOperator.class.getSimpleName());
driverContext.addOperatorContext(5, new PlanNodeId("5"), LimitOperator.class.getSimpleName());
@@ -142,8 +142,8 @@ public class OffsetOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
@@ -159,7 +159,7 @@ public class OffsetOperatorTest {
LimitOperator limitOperator =
new LimitOperator(driverContext.getOperatorContexts().get(4), 250, offsetOperator);
int count = 100;
- while (limitOperator.hasNext()) {
+ while (limitOperator.isBlocked().isDone() && limitOperator.hasNext()) {
TsBlock tsBlock = limitOperator.next();
assertEquals(2, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
@@ -214,7 +214,7 @@ public class OffsetOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
4, new PlanNodeId("4"), OffsetOperator.class.getSimpleName());
driverContext.addOperatorContext(5, new PlanNodeId("5"), LimitOperator.class.getSimpleName());
@@ -247,8 +247,8 @@ public class OffsetOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
@@ -262,7 +262,7 @@ public class OffsetOperatorTest {
new OffsetOperator(driverContext.getOperatorContexts().get(3), 0, timeJoinOperator);
int count = 0;
- while (offsetOperator.hasNext()) {
+ while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
TsBlock tsBlock = offsetOperator.next();
assertEquals(2, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
@@ -316,7 +316,7 @@ public class OffsetOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
4, new PlanNodeId("4"), OffsetOperator.class.getSimpleName());
driverContext.addOperatorContext(5, new PlanNodeId("5"), LimitOperator.class.getSimpleName());
@@ -349,8 +349,8 @@ public class OffsetOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
@@ -363,7 +363,7 @@ public class OffsetOperatorTest {
OffsetOperator offsetOperator =
new OffsetOperator(driverContext.getOperatorContexts().get(3), 500, timeJoinOperator);
- while (offsetOperator.hasNext()) {
+ while (offsetOperator.isBlocked().isDone() && offsetOperator.hasNext()) {
TsBlock tsBlock = offsetOperator.next();
assertEquals(2, tsBlock.getValueColumnCount());
assertTrue(tsBlock.getColumn(0) instanceof IntColumn);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
index e0368f3d70..864bae8d9b 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/OperatorMemoryTest.java
@@ -46,7 +46,6 @@ import org.apache.iotdb.db.mpp.execution.operator.process.SortOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.mpp.execution.operator.process.fill.linear.LinearFill;
import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.TimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.last.AbstractUpdateLastCacheOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.last.LastQueryCollectOperator;
@@ -469,42 +468,6 @@ public class OperatorMemoryTest {
assertEquals(512, sortOperator.calculateRetainedSizeAfterCallingNext());
}
- @Test
- public void timeJoinOperatorTest() {
- List<Operator> children = new ArrayList<>(4);
- List<TSDataType> dataTypeList = new ArrayList<>(2);
- dataTypeList.add(TSDataType.INT32);
- dataTypeList.add(TSDataType.INT32);
- long expectedMaxReturnSize =
- Math.min(
- TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(),
- 3L * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte());
- long expectedMaxPeekMemory = 0;
- long childrenMaxPeekMemory = 0;
-
- for (int i = 0; i < 4; i++) {
- Operator child = Mockito.mock(Operator.class);
- Mockito.when(child.calculateMaxPeekMemory()).thenReturn(128 * 1024L);
- Mockito.when(child.calculateMaxReturnSize()).thenReturn(64 * 1024L);
- Mockito.when(child.calculateRetainedSizeAfterCallingNext()).thenReturn(0L);
- childrenMaxPeekMemory =
- Math.max(childrenMaxPeekMemory, expectedMaxPeekMemory + child.calculateMaxPeekMemory());
- expectedMaxPeekMemory += 64 * 1024L;
- children.add(child);
- }
-
- expectedMaxPeekMemory =
- Math.max(expectedMaxPeekMemory + expectedMaxReturnSize, childrenMaxPeekMemory);
-
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
- Mockito.mock(OperatorContext.class), children, Ordering.ASC, dataTypeList, null, null);
-
- assertEquals(expectedMaxPeekMemory, timeJoinOperator.calculateMaxPeekMemory());
- assertEquals(expectedMaxReturnSize, timeJoinOperator.calculateMaxReturnSize());
- assertEquals(3 * 64 * 1024L, timeJoinOperator.calculateRetainedSizeAfterCallingNext());
- }
-
@Test
public void updateLastCacheOperatorTest() {
Operator child = Mockito.mock(Operator.class);
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
index 083366b9c3..76aa21237e 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/RawDataAggregationOperatorTest.java
@@ -36,7 +36,7 @@ import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.mpp.execution.operator.process.RawDataAggregationOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
@@ -136,7 +136,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -191,7 +192,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -245,7 +247,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(
aggregationTypes, groupByTimeParameter, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -302,7 +305,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(
aggregationTypes, groupByTimeParameter, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -363,7 +367,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(
aggregationTypes, groupByTimeParameter, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -425,7 +430,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(
aggregationTypes, groupByTimeParameter, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -490,7 +496,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(
aggregationTypes, groupByTimeParameter, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -560,7 +567,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -623,7 +631,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -681,7 +690,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -729,7 +739,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -779,7 +790,8 @@ public class RawDataAggregationOperatorTest {
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int resultMinTime1 = -1, resultMinTime2 = -1;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -841,7 +853,8 @@ public class RawDataAggregationOperatorTest {
RawDataAggregationOperator rawDataAggregationOperator =
initRawDataAggregationOperator(aggregationTypes, null, inputLocations, windowParameter);
int count = 0;
- while (rawDataAggregationOperator.hasNext()) {
+ while (rawDataAggregationOperator.isBlocked().isDone()
+ && rawDataAggregationOperator.hasNext()) {
TsBlock resultTsBlock = rawDataAggregationOperator.next();
if (resultTsBlock == null) {
continue;
@@ -891,7 +904,7 @@ public class RawDataAggregationOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
4, new PlanNodeId("4"), RawDataAggregationOperatorTest.class.getSimpleName());
driverContext
@@ -929,8 +942,8 @@ public class RawDataAggregationOperatorTest {
scanOptionsBuilder.build());
seriesScanOperator2.initQueryDataSource(new QueryDataSource(seqResources, unSeqResources));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SingleDeviceViewOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SingleDeviceViewOperatorTest.java
index 7eca183cfe..b935d1d665 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SingleDeviceViewOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/SingleDeviceViewOperatorTest.java
@@ -31,7 +31,7 @@ import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
import org.apache.iotdb.db.mpp.execution.operator.process.SingleDeviceViewOperator;
-import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.SingleColumnMerger;
import org.apache.iotdb.db.mpp.execution.operator.source.SeriesScanOperator;
@@ -146,8 +146,8 @@ public class SingleDeviceViewOperatorTest {
.getOperatorContext()
.setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TimeJoinOperator timeJoinOperator =
- new TimeJoinOperator(
+ RowBasedTimeJoinOperator timeJoinOperator =
+ new RowBasedTimeJoinOperator(
driverContext.getOperatorContexts().get(2),
Arrays.asList(seriesScanOperator1, seriesScanOperator2),
Ordering.ASC,
@@ -165,7 +165,7 @@ public class SingleDeviceViewOperatorTest {
Arrays.asList(TSDataType.TEXT, TSDataType.INT32, TSDataType.INT32, TSDataType.INT32));
int count = 0;
int total = 0;
- while (singleDeviceViewOperator.hasNext()) {
+ while (singleDeviceViewOperator.isBlocked().isDone() && singleDeviceViewOperator.hasNext()) {
TsBlock tsBlock = singleDeviceViewOperator.next();
assertEquals(4, tsBlock.getValueColumnCount());
total += tsBlock.getPositionCount();
diff --git a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/TimeJoinOperatorTest.java b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/TimeJoinOperatorTest.java
index 7737820eb5..70c768c1cf 100644
--- a/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/TimeJoinOperatorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/mpp/execution/operator/TimeJoinOperatorTest.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.mpp.common.QueryId;
import org.apache.iotdb.db.mpp.execution.driver.DriverContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.mpp.execution.fragment.FragmentInstanceStateMachine;
+import org.apache.iotdb.db.mpp.execution.operator.process.join.RowBasedTimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.TimeJoinOperator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.mpp.execution.operator.process.join.merge.DescTimeComparator;
@@ -108,7 +109,7 @@ public class TimeJoinOperatorTest {
PlanNodeId planNodeId2 = new PlanNodeId("2");
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 3, new PlanNodeId("3"), TimeJoinOperator.class.getSimpleName());
+ 3, new PlanNodeId("3"), RowBasedTimeJoinOperator.class.getSimpleName());
SeriesScanOptions.Builder scanOptionsBuilder = new SeriesScanOptions.Builder();
scanOptionsBuilder.withAllSensors(allSensors);