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 2024/07/17 10:36:09 UTC
(iotdb) 01/01: Make Time Column as a normal value column in table model
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch ty/NormalizeTimeColumn
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 25389c72225d4296f31da0e9eb1bae07f7d6a64a
Author: JackieTien97 <ja...@gmail.com>
AuthorDate: Wed Jul 17 18:35:52 2024 +0800
Make Time Column as a normal value column in table model
---
.../src/main/codegen/templates/constantFill.ftl | 3 +-
.../src/main/codegen/templates/previousFill.ftl | 3 +-
.../codegen/templates/variationWindowManager.ftl | 3 +-
.../slidingwindow/SlidingWindowAggregator.java | 7 +-
.../execution/operator/AggregationUtil.java | 3 +-
.../operator/process/AbstractSortOperator.java | 6 +-
.../operator/process/FilterAndProjectOperator.java | 69 +++++--
.../operator/process/MergeSortOperator.java | 12 +-
.../execution/operator/process/SortOperator.java | 11 +-
.../operator/process/TableMergeSortOperator.java | 56 ++++++
.../{SortOperator.java => TableSortOperator.java} | 50 ++---
...tOperator.java => TableStreamSortOperator.java} | 21 ++-
.../operator/process/TableTopKOperator.java | 57 ++++++
.../execution/operator/process/TopKOperator.java | 24 +--
.../operator/process/TransformOperator.java | 11 +-
.../operator/process/TreeMergeSortOperator.java | 53 ++++++
.../{SortOperator.java => TreeSortOperator.java} | 45 +----
.../operator/process/TreeTopKOperator.java | 57 ++++++
.../execution/operator/process/fill/IFill.java | 3 +-
.../operator/process/fill/ILinearFill.java | 5 +-
.../process/fill/identity/IdentityFill.java | 3 +-
.../process/fill/identity/IdentityLinearFill.java | 5 +-
.../operator/process/fill/linear/LinearFill.java | 13 +-
.../process/join/HorizontallyConcatOperator.java | 3 +-
.../process/join/LeftOuterTimeJoinOperator.java | 3 +-
.../process/join/merge/MergeSortComparator.java | 4 +
.../process/join/merge/MultiColumnMerger.java | 5 +-
.../process/join/merge/SingleColumnMerger.java | 3 +-
.../operator/source/AlignedSeriesScanOperator.java | 3 +-
.../operator/source/SeriesScanOperator.java | 3 +-
.../source/relational/TableScanOperator.java | 10 +-
.../operator/window/ConditionWindowManager.java | 3 +-
.../operator/window/CountWindowManager.java | 3 +-
.../operator/window/SessionWindowManager.java | 4 +-
.../plan/optimization/PredicatePushDown.java | 2 +-
.../plan/planner/OperatorTreeGenerator.java | 20 +-
.../plan/planner/TableOperatorGenerator.java | 92 +++-------
.../statistics/SpecifiedInfoMergerFactory.java | 4 +-
.../SingleInputMultiReferenceLayer.java | 14 +-
.../SingleInputSingleReferenceLayer.java | 12 +-
.../iterator/TVListForwardIterator.java | 3 +-
.../tv/ElasticSerializableBinaryTVList.java | 8 +-
.../tv/ElasticSerializableTVList.java | 11 +-
.../datastructure/tv/SerializableTVList.java | 15 +-
.../writer/ReadPointCrossCompactionWriter.java | 2 +-
.../writer/ReadPointInnerCompactionWriter.java | 2 +-
...est.java => LastQueryTreeSortOperatorTest.java} | 4 +-
...torTest.java => MergeTreeSortOperatorTest.java} | 204 ++++++++++++---------
.../execution/operator/OperatorMemoryTest.java | 12 +-
...OperatorTest.java => TreeSortOperatorTest.java} | 24 +--
...OperatorTest.java => TreeTopKOperatorTest.java} | 89 ++++-----
...rTest.java => TableStreamSortOperatorTest.java} | 40 ++--
.../datastructure/SerializableTVListTest.java | 2 +-
53 files changed, 665 insertions(+), 459 deletions(-)
diff --git a/iotdb-core/datanode/src/main/codegen/templates/constantFill.ftl b/iotdb-core/datanode/src/main/codegen/templates/constantFill.ftl
index f29663b47cf..d0523413002 100644
--- a/iotdb-core/datanode/src/main/codegen/templates/constantFill.ftl
+++ b/iotdb-core/datanode/src/main/codegen/templates/constantFill.ftl
@@ -27,7 +27,6 @@ import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.tsfile.read.common.block.column.${type.column};
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
<#if type.dataType == "Binary">
import org.apache.tsfile.utils.Binary;
</#if>
@@ -51,7 +50,7 @@ public class ${className} implements IFill {
}
@Override
- public Column fill(TimeColumn timeColumn, Column valueColumn) {
+ public Column fill(Column timeColumn, Column valueColumn) {
int size = valueColumn.getPositionCount();
// if this valueColumn doesn't have any null value, or it's empty, just return itself;
if (!valueColumn.mayHaveNull() || size == 0) {
diff --git a/iotdb-core/datanode/src/main/codegen/templates/previousFill.ftl b/iotdb-core/datanode/src/main/codegen/templates/previousFill.ftl
index c7f3e73e592..8da4664ead8 100644
--- a/iotdb-core/datanode/src/main/codegen/templates/previousFill.ftl
+++ b/iotdb-core/datanode/src/main/codegen/templates/previousFill.ftl
@@ -30,7 +30,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.read.common.block.column.${type.column};
import org.apache.tsfile.read.common.block.column.${type.column}Builder;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
<#if type.dataType == "Binary">
import org.apache.tsfile.utils.Binary;
</#if>
@@ -57,7 +56,7 @@ public class ${className} implements IFill {
}
@Override
- public Column fill(TimeColumn timeColumn, Column valueColumn) {
+ public Column fill(Column timeColumn, Column valueColumn) {
int size = valueColumn.getPositionCount();
// if this valueColumn is empty, just return itself;
if (size == 0) {
diff --git a/iotdb-core/datanode/src/main/codegen/templates/variationWindowManager.ftl b/iotdb-core/datanode/src/main/codegen/templates/variationWindowManager.ftl
index 6666874b7fd..6ac2e7b8f68 100644
--- a/iotdb-core/datanode/src/main/codegen/templates/variationWindowManager.ftl
+++ b/iotdb-core/datanode/src/main/codegen/templates/variationWindowManager.ftl
@@ -38,7 +38,6 @@ package org.apache.iotdb.db.queryengine.execution.operator.window;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
<#if dataType.dataType == "Binary">
import org.apache.tsfile.utils.Binary;
</#if>
@@ -65,7 +64,7 @@ public class ${className} extends VariationWindowManager {
}
Column controlColumn = variationWindow.getControlColumn(inputTsBlock);
- TimeColumn timeColumn = inputTsBlock.getTimeColumn();
+ Column timeColumn = inputTsBlock.getTimeColumn();
int i = 0, size = inputTsBlock.getPositionCount();
${dataType.dataType} previousValue = ((${windowName}) variationWindow).getPreviousHeadValue();
boolean previousValueIsNull = ((${windowName}) variationWindow).valueIsNull();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregator.java
index 1f6eb9c7925..3421487700e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/slidingwindow/SlidingWindowAggregator.java
@@ -28,7 +28,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import java.util.Arrays;
import java.util.Deque;
@@ -55,7 +54,7 @@ public abstract class SlidingWindowAggregator extends Aggregator {
checkArgument(
step.isInputPartial(),
"Step in SlidingWindowAggregationOperator can only process partial result");
- TimeColumn timeColumn = tsBlock.getTimeColumn();
+ Column timeColumn = tsBlock.getTimeColumn();
Column[] valueColumn = new Column[inputLocationList.get(0).length];
for (int i = 0; i < inputLocationList.get(0).length; i++) {
InputLocation inputLocation = inputLocationList.get(0)[i];
@@ -80,10 +79,10 @@ public abstract class SlidingWindowAggregator extends Aggregator {
protected static class PartialAggregationResult {
- private final TimeColumn timeColumn;
+ private final Column timeColumn;
private final Column[] partialResultColumns;
- public PartialAggregationResult(TimeColumn timeColumn, Column[] partialResultColumns) {
+ public PartialAggregationResult(Column timeColumn, Column[] partialResultColumns) {
this.timeColumn = timeColumn;
this.partialResultColumns = partialResultColumns;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AggregationUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AggregationUtil.java
index 1effb47d890..cbf30889650 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AggregationUtil.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/AggregationUtil.java
@@ -31,6 +31,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationDe
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.GroupByTimeParameter;
import org.apache.iotdb.db.queryengine.statistics.StatisticsManager;
+import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
@@ -127,7 +128,7 @@ public class AggregationUtil {
TsBlock inputTsBlock, TimeRange curTimeRange, List<Aggregator> aggregators) {
// Get the row which need to be processed by aggregator
IWindow curWindow = new TimeWindow(curTimeRange);
- TimeColumn timeColumn = inputTsBlock.getTimeColumn();
+ Column timeColumn = inputTsBlock.getTimeColumn();
int lastIndexToProcess = 0;
for (int i = 0; i < inputTsBlock.getPositionCount(); i++) {
if (!curWindow.satisfy(timeColumn, i)) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractSortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractSortOperator.java
index ca152e6d77b..9f059a6c62c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractSortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AbstractSortOperator.java
@@ -194,7 +194,7 @@ public abstract class AbstractSortOperator implements ProcessOperator {
for (int i = curRow; i < cachedData.size(); i++) {
SortKey sortKey = cachedData.get(i);
TsBlock tsBlock = sortKey.tsBlock;
- timeColumnBuilder.writeLong(tsBlock.getTimeByIndex(sortKey.rowIndex));
+ appendTime(timeColumnBuilder, tsBlock.getTimeByIndex(sortKey.rowIndex));
for (int j = 0; j < valueColumnBuilders.length; j++) {
if (tsBlock.getColumn(j).isNull(sortKey.rowIndex)) {
valueColumnBuilders[j].appendNull();
@@ -225,7 +225,7 @@ public abstract class AbstractSortOperator implements ProcessOperator {
MergeSortKey mergeSortKey = mergeSortHeap.poll();
TsBlock targetBlock = mergeSortKey.tsBlock;
- timeBuilder.writeLong(targetBlock.getTimeByIndex(mergeSortKey.rowIndex));
+ appendTime(timeBuilder, targetBlock.getTimeByIndex(mergeSortKey.rowIndex));
for (int i = 0; i < valueColumnBuilders.length; i++) {
if (targetBlock.getColumn(i).isNull(mergeSortKey.rowIndex)) {
valueColumnBuilders[i].appendNull();
@@ -252,6 +252,8 @@ public abstract class AbstractSortOperator implements ProcessOperator {
sortCost += System.nanoTime() - startTime;
}
+ protected abstract void appendTime(TimeColumnBuilder timeBuilder, long time);
+
private void initMergeSortHeap() throws IoTDBException {
if (mergeSortHeap == null) {
mergeSortHeap = new MergeSortHeap(sortReaders.size(), comparator);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/FilterAndProjectOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/FilterAndProjectOperator.java
index 21bf3d62c6d..4db7dd29431 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/FilterAndProjectOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/FilterAndProjectOperator.java
@@ -39,7 +39,7 @@ import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -132,7 +132,7 @@ public class FilterAndProjectOperator implements ProcessOperator {
* subexpressions after filtering.
*/
private TsBlock getFilterTsBlock(TsBlock input) {
- final TimeColumn originTimeColumn = input.getTimeColumn();
+ final Column originTimeColumn = input.getTimeColumn();
final int positionCount = originTimeColumn.getPositionCount();
// feed Filter ColumnTransformer, including TimeStampColumnTransformer and constant
for (LeafColumnTransformer leafColumnTransformer : filterLeafColumnTransformerList) {
@@ -146,9 +146,6 @@ public class FilterAndProjectOperator implements ProcessOperator {
// reuse this builder
filterTsBlockBuilder.reset();
- final TimeColumnBuilder timeBuilder = filterTsBlockBuilder.getTimeColumnBuilder();
- final ColumnBuilder[] columnBuilders = filterTsBlockBuilder.getValueColumnBuilders();
-
List<Column> resultColumns = new ArrayList<>();
for (int i = 0, n = input.getValueColumnCount(); i < n; i++) {
resultColumns.add(input.getColumn(i));
@@ -161,24 +158,35 @@ public class FilterAndProjectOperator implements ProcessOperator {
}
}
- int rowCount =
- constructFilteredTsBlock(
- resultColumns,
- timeBuilder,
- filterColumn,
- originTimeColumn,
- columnBuilders,
- positionCount);
-
- filterTsBlockBuilder.declarePositions(rowCount);
- return filterTsBlockBuilder.build();
+ final ColumnBuilder[] columnBuilders = filterTsBlockBuilder.getValueColumnBuilders();
+ if (originTimeColumn instanceof RunLengthEncodedColumn) {
+ int rowCount =
+ constructFilteredTsBlock(resultColumns, filterColumn, columnBuilders, positionCount);
+ filterTsBlockBuilder.declarePositions(rowCount);
+
+ return filterTsBlockBuilder.build(originTimeColumn.getRegion(0, rowCount));
+
+ } else {
+ final TimeColumnBuilder timeBuilder = filterTsBlockBuilder.getTimeColumnBuilder();
+ int rowCount =
+ constructFilteredTsBlock(
+ resultColumns,
+ timeBuilder,
+ filterColumn,
+ originTimeColumn,
+ columnBuilders,
+ positionCount);
+
+ filterTsBlockBuilder.declarePositions(rowCount);
+ return filterTsBlockBuilder.build();
+ }
}
private int constructFilteredTsBlock(
List<Column> resultColumns,
TimeColumnBuilder timeBuilder,
Column filterColumn,
- TimeColumn originTimeColumn,
+ Column originTimeColumn,
ColumnBuilder[] columnBuilders,
int positionCount) {
// construct result TsBlock of filter
@@ -202,12 +210,37 @@ public class FilterAndProjectOperator implements ProcessOperator {
return rowCount;
}
+ private int constructFilteredTsBlock(
+ List<Column> resultColumns,
+ Column filterColumn,
+ ColumnBuilder[] columnBuilders,
+ int positionCount) {
+ // construct result TsBlock of filter
+ int rowCount = 0;
+ for (int i = 0, n = resultColumns.size(); i < n; i++) {
+ Column curColumn = resultColumns.get(i);
+ for (int j = 0; j < positionCount; j++) {
+ if (satisfy(filterColumn, j)) {
+ if (i == 0) {
+ rowCount++;
+ }
+ if (curColumn.isNull(j)) {
+ columnBuilders[i].appendNull();
+ } else {
+ columnBuilders[i].write(curColumn, j);
+ }
+ }
+ }
+ }
+ return rowCount;
+ }
+
private boolean satisfy(Column filterColumn, int rowIndex) {
return !filterColumn.isNull(rowIndex) && filterColumn.getBoolean(rowIndex);
}
private TsBlock getTransformedTsBlock(TsBlock input) {
- final TimeColumn originTimeColumn = input.getTimeColumn();
+ final Column originTimeColumn = input.getTimeColumn();
final int positionCount = originTimeColumn.getPositionCount();
// feed pre calculated data
for (LeafColumnTransformer leafColumnTransformer : projectLeafColumnTransformerList) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/MergeSortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/MergeSortOperator.java
index a80219c9975..813217872ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/MergeSortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/MergeSortOperator.java
@@ -42,7 +42,7 @@ import java.util.concurrent.TimeUnit;
import static com.google.common.util.concurrent.Futures.successfulAsList;
-public class MergeSortOperator extends AbstractConsumeAllOperator {
+public abstract class MergeSortOperator extends AbstractConsumeAllOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(MergeSortOperator.class);
@@ -55,7 +55,7 @@ public class MergeSortOperator extends AbstractConsumeAllOperator {
private boolean finished;
- public MergeSortOperator(
+ MergeSortOperator(
OperatorContext operatorContext,
List<Operator> inputOperators,
List<TSDataType> dataTypes,
@@ -125,7 +125,7 @@ public class MergeSortOperator extends AbstractConsumeAllOperator {
MergeSortKey mergeSortKey = mergeSortHeap.poll();
TsBlock targetBlock = mergeSortKey.tsBlock;
int rowIndex = mergeSortKey.rowIndex;
- timeBuilder.writeLong(targetBlock.getTimeByIndex(rowIndex));
+ appendTime(timeBuilder, targetBlock.getTimeByIndex(rowIndex));
for (int i = 0; i < valueColumnBuilders.length; i++) {
if (targetBlock.getColumn(i).isNull(rowIndex)) {
valueColumnBuilders[i].appendNull();
@@ -146,9 +146,13 @@ public class MergeSortOperator extends AbstractConsumeAllOperator {
break;
}
}
- return tsBlockBuilder.build();
+ return buildResult(tsBlockBuilder);
}
+ protected abstract void appendTime(TimeColumnBuilder timeBuilder, long time);
+
+ protected abstract TsBlock buildResult(TsBlockBuilder resultBuilder);
+
@Override
public boolean hasNext() throws Exception {
if (finished) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
index 0f9ca2bd848..44fae368bca 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -27,17 +27,18 @@ import org.apache.iotdb.db.utils.datastructure.SortKey;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
import java.util.Comparator;
import java.util.List;
-public class SortOperator extends AbstractSortOperator {
+public abstract class SortOperator extends AbstractSortOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(SortOperator.class);
- public SortOperator(
+ SortOperator(
OperatorContext operatorContext,
Operator inputOperator,
List<TSDataType> dataTypes,
@@ -50,7 +51,7 @@ public class SortOperator extends AbstractSortOperator {
public TsBlock next() throws Exception {
if (!inputOperator.hasNextWithTimer()) {
buildResult();
- TsBlock res = tsBlockBuilder.build();
+ TsBlock res = buildFinalResult(tsBlockBuilder);
tsBlockBuilder.reset();
return res;
}
@@ -72,6 +73,8 @@ public class SortOperator extends AbstractSortOperator {
return null;
}
+ protected abstract TsBlock buildFinalResult(TsBlockBuilder resultBuilder);
+
@Override
public long ramBytesUsed() {
return INSTANCE_SIZE
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableMergeSortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableMergeSortOperator.java
new file mode 100644
index 00000000000..0a7616db38e
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableMergeSortOperator.java
@@ -0,0 +1,56 @@
+/*
+ * 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.queryengine.execution.operator.process;
+
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
+
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public class TableMergeSortOperator extends MergeSortOperator {
+ public TableMergeSortOperator(
+ OperatorContext operatorContext,
+ List<Operator> inputOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator) {
+ super(operatorContext, inputOperators, dataTypes, comparator);
+ }
+
+ @Override
+ protected void appendTime(TimeColumnBuilder timeBuilder, long time) {
+ // do nothing for table related operator
+ }
+
+ @Override
+ protected TsBlock buildResult(TsBlockBuilder resultBuilder) {
+ return resultBuilder.build(
+ new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, resultBuilder.getPositionCount()));
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableSortOperator.java
similarity index 51%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableSortOperator.java
index 0f9ca2bd848..208c0327996 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableSortOperator.java
@@ -7,7 +7,7 @@
* "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
+ * 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
@@ -19,25 +19,23 @@
package org.apache.iotdb.db.queryengine.execution.operator.process;
-import org.apache.iotdb.commons.exception.IoTDBException;
-import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
import org.apache.iotdb.db.queryengine.execution.operator.Operator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.utils.datastructure.SortKey;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.utils.RamUsageEstimator;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import java.util.Comparator;
import java.util.List;
-public class SortOperator extends AbstractSortOperator {
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
- private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(SortOperator.class);
-
- public SortOperator(
+public class TableSortOperator extends SortOperator {
+ public TableSortOperator(
OperatorContext operatorContext,
Operator inputOperator,
List<TSDataType> dataTypes,
@@ -47,37 +45,13 @@ public class SortOperator extends AbstractSortOperator {
}
@Override
- public TsBlock next() throws Exception {
- if (!inputOperator.hasNextWithTimer()) {
- buildResult();
- TsBlock res = tsBlockBuilder.build();
- tsBlockBuilder.reset();
- return res;
- }
- long startTime = System.nanoTime();
- try {
- TsBlock tsBlock = inputOperator.nextWithTimer();
- if (tsBlock == null) {
- return null;
- }
- dataSize += tsBlock.getRetainedSizeInBytes();
- cacheTsBlock(tsBlock);
- } catch (IoTDBException e) {
- clear();
- throw e;
- } finally {
- prepareUntilReadyCost += System.nanoTime() - startTime;
- }
-
- return null;
+ protected void appendTime(TimeColumnBuilder timeBuilder, long time) {
+ // do nothing for table related operator
}
@Override
- public long ramBytesUsed() {
- return INSTANCE_SIZE
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(inputOperator)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
- + RamUsageEstimator.sizeOf(noMoreData)
- + tsBlockBuilder.getRetainedSizeInBytes();
+ protected TsBlock buildFinalResult(TsBlockBuilder resultBuilder) {
+ return resultBuilder.build(
+ new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, resultBuilder.getPositionCount()));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperator.java
similarity index 89%
rename from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperator.java
rename to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperator.java
index d53a3480f93..eb7025fcb30 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperator.java
@@ -27,15 +27,19 @@ import org.apache.iotdb.db.utils.datastructure.SortKey;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
import java.util.Comparator;
import java.util.List;
-public class StreamSortOperator extends AbstractSortOperator {
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public class TableStreamSortOperator extends AbstractSortOperator {
private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(StreamSortOperator.class);
+ RamUsageEstimator.shallowSizeOfInstance(TableStreamSortOperator.class);
private final Comparator<SortKey> streamSortComparator;
@@ -55,7 +59,7 @@ public class StreamSortOperator extends AbstractSortOperator {
private SortKey lastRow = null;
- public StreamSortOperator(
+ public TableStreamSortOperator(
OperatorContext operatorContext,
Operator inputOperator,
List<TSDataType> dataTypes,
@@ -82,8 +86,10 @@ public class StreamSortOperator extends AbstractSortOperator {
}
if (tsBlockBuilder.isFull() || consumedUp()) {
- TsBlock res = tsBlockBuilder.build();
- remainingCount -= res.getPositionCount();
+ int rowCount = tsBlockBuilder.getPositionCount();
+ TsBlock res =
+ tsBlockBuilder.build(new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, rowCount));
+ remainingCount -= rowCount;
tsBlockBuilder.reset();
return res;
}
@@ -180,6 +186,11 @@ public class StreamSortOperator extends AbstractSortOperator {
lastRow = new SortKey(tsBlock, tsBlock.getPositionCount() - 1);
}
+ @Override
+ protected void appendTime(TimeColumnBuilder timeBuilder, long time) {
+ // do nothing for Table related Operator
+ }
+
@Override
public boolean hasNext() throws Exception {
return super.hasNext() || !tsBlockBuilder.isEmpty() || currentTsBlock != null;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableTopKOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableTopKOperator.java
new file mode 100644
index 00000000000..ca4a5ad5beb
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableTopKOperator.java
@@ -0,0 +1,57 @@
+/*
+ * 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.queryengine.execution.operator.process;
+
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+
+import java.util.Comparator;
+import java.util.List;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public class TableTopKOperator extends TopKOperator {
+ public TableTopKOperator(
+ OperatorContext operatorContext,
+ List<Operator> childrenOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator,
+ int topValue,
+ boolean childrenDataInOrder) {
+ super(operatorContext, childrenOperators, dataTypes, comparator, topValue, childrenDataInOrder);
+ }
+
+ @Override
+ protected TsBlock constrcutResultTsBlock(int positionCount, Column[] columns) {
+ return new TsBlock(new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, positionCount), columns);
+ }
+
+ @Override
+ protected void updateTsBlock(
+ TsBlock resultTsBlock, int updateIdx, TsBlock sourceTsBlock, int sourceIndex) {
+ resultTsBlock.updateWithoutTimeColumn(updateIdx, sourceTsBlock, sourceIndex);
+ }
+}
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 ad4e04e2aed..8cd8f94fea6 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
@@ -39,7 +39,6 @@ import org.apache.tsfile.read.common.block.column.DoubleColumn;
import org.apache.tsfile.read.common.block.column.FloatColumn;
import org.apache.tsfile.read.common.block.column.IntColumn;
import org.apache.tsfile.read.common.block.column.LongColumn;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.UnSupportedDataTypeException;
@@ -52,10 +51,10 @@ import java.util.concurrent.TimeUnit;
import static com.google.common.util.concurrent.Futures.successfulAsList;
-public class TopKOperator implements ProcessOperator {
+public abstract class TopKOperator implements ProcessOperator {
private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(TopKOperator.class);
+ RamUsageEstimator.shallowSizeOfInstance(TreeTopKOperator.class);
private final OperatorContext operatorContext;
private final List<Operator> childrenOperators;
@@ -67,13 +66,13 @@ public class TopKOperator implements ProcessOperator {
private final List<TSDataType> dataTypes;
private final TsBlockBuilder tsBlockBuilder;
- // max heap, revered of MergeSortHeap in MergeSortOperator
+ // max heap, revered of MergeSortHeap in TreeMergeSortOperator
private final MergeSortHeap mergeSortHeap;
private final Comparator<SortKey> comparator;
// value in LIMIT
private final int topValue;
- // final query result of TopKOperator and the returned size of topKResult
+ // final query result of TreeTopKOperator and the returned size of topKResult
private MergeSortKey[] topKResult;
private int resultReturnSize = 0;
@@ -87,7 +86,7 @@ public class TopKOperator implements ProcessOperator {
public static final int OPERATOR_BATCH_UPPER_BOUND = 100000;
- public TopKOperator(
+ TopKOperator(
OperatorContext operatorContext,
List<Operator> childrenOperators,
List<TSDataType> dataTypes,
@@ -313,10 +312,11 @@ public class TopKOperator implements ProcessOperator {
throw new UnSupportedDataTypeException("Unknown datatype: " + dataTypes.get(i));
}
}
- this.tmpResultTsBlock =
- new TsBlock(positionCount, new TimeColumn(positionCount, new long[positionCount]), columns);
+ this.tmpResultTsBlock = constrcutResultTsBlock(positionCount, columns);
}
+ protected abstract TsBlock constrcutResultTsBlock(int positionCount, Column[] columns);
+
private TsBlock getResultFromCachedTopKResult() {
if (mergeSortHeap.getHeapSize() > 0) {
int cnt = mergeSortHeap.getHeapSize();
@@ -385,15 +385,17 @@ public class TopKOperator implements ProcessOperator {
private void updateTsBlockValue(TsBlock sourceTsBlock, int sourceIndex, int peekIndex) {
if (peekIndex < 0) {
- tmpResultTsBlock.update(tmpResultTsBlockIdx, sourceTsBlock, sourceIndex);
+ updateTsBlock(tmpResultTsBlock, tmpResultTsBlockIdx, sourceTsBlock, sourceIndex);
mergeSortHeap.push(new MergeSortKey(tmpResultTsBlock, tmpResultTsBlockIdx++));
return;
}
-
- tmpResultTsBlock.update(peekIndex, sourceTsBlock, sourceIndex);
+ updateTsBlock(tmpResultTsBlock, peekIndex, sourceTsBlock, sourceIndex);
mergeSortHeap.push(new MergeSortKey(tmpResultTsBlock, peekIndex));
}
+ protected abstract void updateTsBlock(
+ TsBlock resultTsBlock, int updateIdx, TsBlock sourceTsBlock, int sourceIndex);
+
private Operator getOperator(int i) {
return childrenOperators.get(i);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TransformOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TransformOperator.java
index be418dba049..96c7aaeaeb9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TransformOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TransformOperator.java
@@ -45,7 +45,6 @@ import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.UnSupportedDataTypeException;
@@ -181,11 +180,7 @@ public class TransformOperator implements ProcessOperator {
}
Column[] columns = transformers[index].current();
-
- int count = columns[0].getPositionCount();
- Column[] valueColumns = {columns[0]};
- TsBlock.wrapBlocksWithoutCopy(count, (TimeColumn) columns[1], valueColumns);
- TsBlock block = new TsBlock((TimeColumn) columns[1], columns[0]);
+ TsBlock block = new TsBlock(columns[1], columns[0]);
outputColumns[index] = block;
currentIndexes[index] = 0;
}
@@ -202,7 +197,7 @@ public class TransformOperator implements ProcessOperator {
}
if (outputColumns[index] != null) {
- TimeColumn outputTimeColumn = outputColumns[index].getTimeColumn();
+ Column outputTimeColumn = outputColumns[index].getTimeColumn();
long time = outputTimeColumn.getLong(currentIndexes[index]);
timeHeap.add(time);
return YieldableState.YIELDABLE;
@@ -317,7 +312,7 @@ public class TransformOperator implements ProcessOperator {
return state;
}
- TimeColumn timeColumn = outputColumns[index].getTimeColumn();
+ Column timeColumn = outputColumns[index].getTimeColumn();
Column valueColumn = outputColumns[index].getColumn(0);
int currentIndex = currentIndexes[index];
if (timeColumn.getLong(currentIndex) != currentTime) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeMergeSortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeMergeSortOperator.java
new file mode 100644
index 00000000000..c5ba949ad37
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeMergeSortOperator.java
@@ -0,0 +1,53 @@
+/*
+ * 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.queryengine.execution.operator.process;
+
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class TreeMergeSortOperator extends MergeSortOperator {
+
+ public TreeMergeSortOperator(
+ OperatorContext operatorContext,
+ List<Operator> inputOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator) {
+ super(operatorContext, inputOperators, dataTypes, comparator);
+ }
+
+ @Override
+ protected void appendTime(TimeColumnBuilder timeBuilder, long time) {
+ timeBuilder.writeLong(time);
+ }
+
+ @Override
+ protected TsBlock buildResult(TsBlockBuilder resultBuilder) {
+ return resultBuilder.build();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeSortOperator.java
similarity index 53%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeSortOperator.java
index 0f9ca2bd848..2e3a4801c22 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/SortOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeSortOperator.java
@@ -19,25 +19,21 @@
package org.apache.iotdb.db.queryengine.execution.operator.process;
-import org.apache.iotdb.commons.exception.IoTDBException;
-import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
import org.apache.iotdb.db.queryengine.execution.operator.Operator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.utils.datastructure.SortKey;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.utils.RamUsageEstimator;
+import org.apache.tsfile.read.common.block.TsBlockBuilder;
+import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import java.util.Comparator;
import java.util.List;
-public class SortOperator extends AbstractSortOperator {
+public class TreeSortOperator extends SortOperator {
- private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(SortOperator.class);
-
- public SortOperator(
+ public TreeSortOperator(
OperatorContext operatorContext,
Operator inputOperator,
List<TSDataType> dataTypes,
@@ -47,37 +43,12 @@ public class SortOperator extends AbstractSortOperator {
}
@Override
- public TsBlock next() throws Exception {
- if (!inputOperator.hasNextWithTimer()) {
- buildResult();
- TsBlock res = tsBlockBuilder.build();
- tsBlockBuilder.reset();
- return res;
- }
- long startTime = System.nanoTime();
- try {
- TsBlock tsBlock = inputOperator.nextWithTimer();
- if (tsBlock == null) {
- return null;
- }
- dataSize += tsBlock.getRetainedSizeInBytes();
- cacheTsBlock(tsBlock);
- } catch (IoTDBException e) {
- clear();
- throw e;
- } finally {
- prepareUntilReadyCost += System.nanoTime() - startTime;
- }
-
- return null;
+ protected TsBlock buildFinalResult(TsBlockBuilder resultBuilder) {
+ return resultBuilder.build();
}
@Override
- public long ramBytesUsed() {
- return INSTANCE_SIZE
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(inputOperator)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
- + RamUsageEstimator.sizeOf(noMoreData)
- + tsBlockBuilder.getRetainedSizeInBytes();
+ protected void appendTime(TimeColumnBuilder timeBuilder, long time) {
+ timeBuilder.writeLong(time);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeTopKOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeTopKOperator.java
new file mode 100644
index 00000000000..bef0b47819e
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/TreeTopKOperator.java
@@ -0,0 +1,57 @@
+/*
+ * 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.queryengine.execution.operator.process;
+
+import org.apache.iotdb.db.queryengine.execution.operator.Operator;
+import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
+import org.apache.iotdb.db.utils.datastructure.SortKey;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.TimeColumn;
+
+import java.util.Comparator;
+import java.util.List;
+
+public class TreeTopKOperator extends TopKOperator {
+
+ public TreeTopKOperator(
+ OperatorContext operatorContext,
+ List<Operator> childrenOperators,
+ List<TSDataType> dataTypes,
+ Comparator<SortKey> comparator,
+ int topValue,
+ boolean childrenDataInOrder) {
+ super(operatorContext, childrenOperators, dataTypes, comparator, topValue, childrenDataInOrder);
+ }
+
+ @Override
+ protected TsBlock constrcutResultTsBlock(int positionCount, Column[] columns) {
+ return new TsBlock(
+ positionCount, new TimeColumn(positionCount, new long[positionCount]), columns);
+ }
+
+ @Override
+ protected void updateTsBlock(
+ TsBlock resultTsBlock, int updateIdx, TsBlock sourceTsBlock, int sourceIndex) {
+ resultTsBlock.update(updateIdx, sourceTsBlock, sourceIndex);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/IFill.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/IFill.java
index e936db864f5..62c947fa4cd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/IFill.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/IFill.java
@@ -20,9 +20,8 @@
package org.apache.iotdb.db.queryengine.execution.operator.process.fill;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
public interface IFill {
- Column fill(TimeColumn timeColumn, Column valueColumn);
+ Column fill(Column timeColumn, Column valueColumn);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/ILinearFill.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/ILinearFill.java
index 101be3e9a7d..c20a120901b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/ILinearFill.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/ILinearFill.java
@@ -20,7 +20,6 @@
package org.apache.iotdb.db.queryengine.execution.operator.process.fill;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
public interface ILinearFill {
@@ -33,7 +32,7 @@ public interface ILinearFill {
* @param currentRowIndex current row index for start time in timeColumn
* @return Value Column that has been filled
*/
- Column fill(TimeColumn timeColumn, Column valueColumn, long currentRowIndex);
+ Column fill(Column timeColumn, Column valueColumn, long currentRowIndex);
/**
* Whether need prepare for next.
@@ -58,5 +57,5 @@ public interface ILinearFill {
* current column, and still need to keep getting next TsBlock and then call prepareForNext
*/
boolean prepareForNext(
- long startRowIndex, long endRowIndex, TimeColumn nextTimeColumn, Column nextValueColumn);
+ long startRowIndex, long endRowIndex, Column nextTimeColumn, Column nextValueColumn);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityFill.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityFill.java
index 6dcdb6ff8a2..df0a932d7cd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityFill.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityFill.java
@@ -22,12 +22,11 @@ package org.apache.iotdb.db.queryengine.execution.operator.process.fill.identity
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
public class IdentityFill implements IFill {
@Override
- public Column fill(TimeColumn timeColumn, Column valueColumn) {
+ public Column fill(Column timeColumn, Column valueColumn) {
return valueColumn;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityLinearFill.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityLinearFill.java
index 7396a31ad19..3bd579d53d7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityLinearFill.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/identity/IdentityLinearFill.java
@@ -22,12 +22,11 @@ package org.apache.iotdb.db.queryengine.execution.operator.process.fill.identity
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.ILinearFill;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
public class IdentityLinearFill implements ILinearFill {
@Override
- public Column fill(TimeColumn timeColumn, Column valueColumn, long currentRowIndex) {
+ public Column fill(Column timeColumn, Column valueColumn, long currentRowIndex) {
return valueColumn;
}
@@ -38,7 +37,7 @@ public class IdentityLinearFill implements ILinearFill {
@Override
public boolean prepareForNext(
- long startRowIndex, long endRowIndex, TimeColumn nextTimeColumn, Column nextValueColumn) {
+ long startRowIndex, long endRowIndex, Column nextTimeColumn, Column nextValueColumn) {
throw new UnsupportedOperationException(
"IdentityLinearFill's needPrepareForNext() method should always return false.");
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/linear/LinearFill.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/linear/LinearFill.java
index d35ec87d01b..9a62d10c006 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/linear/LinearFill.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/fill/linear/LinearFill.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.db.queryengine.execution.operator.process.fill.ILinearFi
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import static com.google.common.base.Preconditions.checkArgument;
@@ -51,7 +50,7 @@ public abstract class LinearFill implements ILinearFill {
protected long nextTimeInCurrentColumn = -1;
@Override
- public Column fill(TimeColumn timeColumn, Column valueColumn, long startRowIndex) {
+ public Column fill(Column timeColumn, Column valueColumn, long startRowIndex) {
int size = valueColumn.getPositionCount();
if (size == 0) {
return valueColumn;
@@ -61,7 +60,7 @@ public abstract class LinearFill implements ILinearFill {
if (!valueColumn.mayHaveNull()) {
previousIsNull = false;
// update the value using last non-null value
- previousTime = timeColumn.getEndTime();
+ previousTime = timeColumn.getLong(timeColumn.getPositionCount() - 1);
updatePreviousValue(valueColumn, valueColumn.getPositionCount() - 1);
return valueColumn;
}
@@ -94,7 +93,7 @@ public abstract class LinearFill implements ILinearFill {
}
private Column doWithAllNulls(
- long startRowIndex, int size, TimeColumn timeColumn, Column valueColumn) {
+ long startRowIndex, int size, Column timeColumn, Column valueColumn) {
// previous value is null or next value is null, we just return NULL_VALUE_BLOCK
if (previousIsNull || nextRowIndex < startRowIndex) {
return new RunLengthEncodedColumn(createNullValueColumn(), size);
@@ -116,7 +115,7 @@ public abstract class LinearFill implements ILinearFill {
long startRowIndex,
int i,
boolean[] isNull,
- TimeColumn timeColumn,
+ Column timeColumn,
Column valueColumn,
Object array) {
long currentRowIndex = startRowIndex + i;
@@ -156,7 +155,7 @@ public abstract class LinearFill implements ILinearFill {
@Override
public boolean prepareForNext(
- long startRowIndex, long endRowIndex, TimeColumn nextTimeColumn, Column nextValueColumn) {
+ long startRowIndex, long endRowIndex, Column nextTimeColumn, Column nextValueColumn) {
checkArgument(
nextTimeColumn.getPositionCount() > 0 && endRowIndex < startRowIndex,
"nextColumn's time should be greater than current time");
@@ -180,7 +179,7 @@ public abstract class LinearFill implements ILinearFill {
}
private void prepareForNextValueInCurrentColumn(
- long currentRowIndex, int startIndex, TimeColumn timeColumn, Column valueColumn) {
+ long currentRowIndex, int startIndex, Column timeColumn, Column valueColumn) {
if (currentRowIndex <= nextRowIndexInCurrentColumn) {
return;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/HorizontallyConcatOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/HorizontallyConcatOperator.java
index c9bcff17fab..6fae539d2f8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/HorizontallyConcatOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/HorizontallyConcatOperator.java
@@ -29,7 +29,6 @@ import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -78,7 +77,7 @@ public class HorizontallyConcatOperator extends AbstractConsumeAllOperator {
Math.min(maxRowCanBuild, inputTsBlocks[i].getPositionCount() - inputIndex[i]);
}
- TimeColumn firstTimeColumn = inputTsBlocks[0].getTimeColumn();
+ Column firstTimeColumn = inputTsBlocks[0].getTimeColumn();
TimeColumnBuilder timeColumnBuilder = tsBlockBuilder.getTimeColumnBuilder();
ColumnBuilder[] valueColumnBuilders = tsBlockBuilder.getValueColumnBuilders();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/LeftOuterTimeJoinOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/LeftOuterTimeJoinOperator.java
index 451a4db328b..0607d0a1e35 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/LeftOuterTimeJoinOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/LeftOuterTimeJoinOperator.java
@@ -32,7 +32,6 @@ import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -254,7 +253,7 @@ public class LeftOuterTimeJoinOperator implements ProcessOperator {
int rowSize = leftTsBlock.getPositionCount();
// append time column
TimeColumnBuilder timeColumnBuilder = resultBuilder.getTimeColumnBuilder();
- TimeColumn leftTimeColumn = leftTsBlock.getTimeColumn();
+ Column leftTimeColumn = leftTsBlock.getTimeColumn();
for (int i = leftIndex; i < rowSize; i++) {
timeColumnBuilder.writeLong(leftTimeColumn.getLong(i));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MergeSortComparator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MergeSortComparator.java
index db6f36206f9..e332e6b7b0b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MergeSortComparator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MergeSortComparator.java
@@ -90,11 +90,13 @@ public class MergeSortComparator {
Comparator<SortKey> comparator;
switch (dataType) {
case INT32:
+ case DATE:
comparator =
Comparator.comparingInt(
(SortKey sortKey) -> sortKey.tsBlock.getColumn(index).getInt(sortKey.rowIndex));
break;
case INT64:
+ case TIMESTAMP:
comparator =
Comparator.comparingLong(
(SortKey sortKey) -> sortKey.tsBlock.getColumn(index).getLong(sortKey.rowIndex));
@@ -110,6 +112,8 @@ public class MergeSortComparator {
(SortKey sortKey) -> sortKey.tsBlock.getColumn(index).getDouble(sortKey.rowIndex));
break;
case TEXT:
+ case BLOB:
+ case STRING:
comparator =
Comparator.comparing(
(SortKey sortKey) -> sortKey.tsBlock.getColumn(index).getBinary(sortKey.rowIndex));
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MultiColumnMerger.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MultiColumnMerger.java
index 16328bf2294..149667ed250 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MultiColumnMerger.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/MultiColumnMerger.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import java.util.List;
@@ -69,7 +68,7 @@ public class MultiColumnMerger implements ColumnMerger {
// current location's input column is not empty
if (!ColumnMerger.empty(tsBlockIndex, inputTsBlocks, updatedInputIndex)) {
- TimeColumn timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
+ Column timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
Column valueColumn = inputTsBlocks[tsBlockIndex].getColumn(columnIndex);
// time of current location's input column is equal to current row's time
if (timeColumn.getLong(index) == timeBuilder.getTime(i)) {
@@ -124,7 +123,7 @@ public class MultiColumnMerger implements ColumnMerger {
// current location's input column is not empty
if (!ColumnMerger.empty(tsBlockIndex, inputTsBlocks, updatedInputIndex)) {
- TimeColumn timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
+ Column timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
Column valueColumn = inputTsBlocks[tsBlockIndex].getColumn(columnIndex);
// time of current location's input column is equal to current row's time
if (timeColumn.getLong(index) == currentTime) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/SingleColumnMerger.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/SingleColumnMerger.java
index 0541669eab2..ba33e219d06 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/SingleColumnMerger.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/join/merge/SingleColumnMerger.java
@@ -24,7 +24,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
/** only has one input column. */
@@ -93,7 +92,7 @@ public class SingleColumnMerger implements ColumnMerger {
columnBuilder.appendNull(rowCount);
} else {
// read from input column and write it into columnBuilder
- TimeColumn timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
+ Column timeColumn = inputTsBlocks[tsBlockIndex].getTimeColumn();
Column valueColumn = inputTsBlocks[tsBlockIndex].getColumn(columnIndex);
for (int i = 0; i < rowCount; i++) {
// current index reaches the size of input column or current time of input column is already
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java
index 733c2a82528..1fc8014b3ea 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanOperator.java
@@ -34,7 +34,6 @@ import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -90,7 +89,7 @@ public class AlignedSeriesScanOperator extends AbstractSeriesScanOperator {
public static void appendDataIntoBuilder(TsBlock tsBlock, TsBlockBuilder builder) {
int size = tsBlock.getPositionCount();
TimeColumnBuilder timeColumnBuilder = builder.getTimeColumnBuilder();
- TimeColumn timeColumn = tsBlock.getTimeColumn();
+ Column timeColumn = tsBlock.getTimeColumn();
for (int i = 0; i < size; i++) {
timeColumnBuilder.writeLong(timeColumn.getLong(i));
builder.declarePosition();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java
index 0a88e711084..f83623a4cc2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanOperator.java
@@ -30,7 +30,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.block.column.ColumnBuilder;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TimeColumnBuilder;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -61,7 +60,7 @@ public class SeriesScanOperator extends AbstractSeriesScanOperator {
protected void buildResult(TsBlock tsBlock) {
int size = tsBlock.getPositionCount();
TimeColumnBuilder timeColumnBuilder = resultTsBlockBuilder.getTimeColumnBuilder();
- TimeColumn timeColumn = tsBlock.getTimeColumn();
+ Column timeColumn = tsBlock.getTimeColumn();
ColumnBuilder columnBuilder = resultTsBlockBuilder.getColumnBuilder(0);
Column column = tsBlock.getColumn(0);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java
index 7ceefaa9310..673722c6de3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java
@@ -40,6 +40,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
import org.apache.tsfile.read.common.block.column.BinaryColumn;
+import org.apache.tsfile.read.common.block.column.LongColumn;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.RamUsageEstimator;
@@ -60,6 +61,9 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(TableScanOperator.class);
+ public static final LongColumn TIME_COLUMN_TEMPLATE =
+ new LongColumn(1, Optional.empty(), new long[] {0});
+
private final List<ColumnSchema> columnSchemas;
private final int[] columnsIndexArray;
@@ -220,6 +224,7 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
valueColumns[i] = measurementDataBlock.getColumn(columnsIndexArray[i]);
break;
case TIME:
+ valueColumns[i] = measurementDataBlock.getTimeColumn();
break;
default:
throw new IllegalArgumentException(
@@ -227,7 +232,10 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
}
}
this.resultTsBlock =
- new TsBlock(positionCount, measurementDataBlock.getTimeColumn(), valueColumns);
+ new TsBlock(
+ positionCount,
+ new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, positionCount),
+ valueColumns);
}
private RunLengthEncodedColumn getIdOrAttributeValueColumn(String value, int positionCount) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/ConditionWindowManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/ConditionWindowManager.java
index 45400e82b92..70d251ef461 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/ConditionWindowManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/ConditionWindowManager.java
@@ -27,7 +27,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import java.util.List;
@@ -126,7 +125,7 @@ public class ConditionWindowManager implements IWindowManager {
}
Column controlColumn = conditionWindow.getControlColumn(inputTsBlock);
- TimeColumn timeColumn = inputTsBlock.getTimeColumn();
+ Column timeColumn = inputTsBlock.getTimeColumn();
int i = 0;
int k = 0;
int size = inputTsBlock.getPositionCount();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/CountWindowManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/CountWindowManager.java
index ac3da9e1d4e..3b86e9b462c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/CountWindowManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/CountWindowManager.java
@@ -25,7 +25,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import java.util.List;
@@ -80,7 +79,7 @@ public class CountWindowManager implements IWindowManager {
return inputTsBlock;
}
- TimeColumn timeColumn = inputTsBlock.getTimeColumn();
+ Column timeColumn = inputTsBlock.getTimeColumn();
Column controlColumn = countWindow.getControlColumn(inputTsBlock);
long leftCount = countWindow.getLeftCount();
int i = 0;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/SessionWindowManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/SessionWindowManager.java
index a5504a8b135..cae70428af2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/SessionWindowManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/window/SessionWindowManager.java
@@ -21,10 +21,10 @@ package org.apache.iotdb.db.queryengine.execution.operator.window;
import org.apache.iotdb.db.queryengine.execution.aggregation.Aggregator;
+import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
import org.apache.tsfile.read.common.block.TsBlockBuilder;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import java.util.List;
@@ -86,7 +86,7 @@ public class SessionWindowManager implements IWindowManager {
return inputTsBlock;
}
- TimeColumn timeColumn = inputTsBlock.getTimeColumn();
+ Column timeColumn = inputTsBlock.getTimeColumn();
int i = 0;
int size = inputTsBlock.getPositionCount();
long previousTimeValue = sessionWindow.getTimeValue();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/PredicatePushDown.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/PredicatePushDown.java
index 65a7023bb31..ff1d4a07ea7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/PredicatePushDown.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/optimization/PredicatePushDown.java
@@ -369,7 +369,7 @@ public class PredicatePushDown implements PlanOptimizer {
* <li>Because of the removal of the FilterNode (FilterAndProjectOperator), we need a
* ProjectNode to do the projection.
* <li>For ALIGN_BY_DEVICE query, the ProjectNode is used to ensure the order of the output
- * columns is consistent with before optimization (required by MergeSortOperator).
+ * columns is consistent with before optimization (required by TreeMergeSortOperator).
*/
private PlanNode planProject(PlanNode resultNode, RewriterContext context) {
FilterNode pushDownFilterNode = context.getPushDownFilterNode();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
index 71dc05861d4..b75a52835f6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/OperatorTreeGenerator.java
@@ -64,17 +64,17 @@ import org.apache.iotdb.db.queryengine.execution.operator.process.FilterAndProje
import org.apache.iotdb.db.queryengine.execution.operator.process.IntoOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.LimitOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.LinearFillOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.OffsetOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.ProcessOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.ProjectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.RawDataAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SingleDeviceViewOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SlidingWindowAggregationOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TagAggregationOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.TransformOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeMergeSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeTopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFillFilter;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.ILinearFill;
@@ -1142,7 +1142,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- MergeSortOperator.class.getSimpleName());
+ TreeMergeSortOperator.class.getSimpleName());
List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
context.setCachedDataTypes(dataTypes);
List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
@@ -1156,7 +1156,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
sortItemList,
sortItemIndexList,
sortItemDataTypeList);
- return new MergeSortOperator(
+ return new TreeMergeSortOperator(
operatorContext,
children,
dataTypes,
@@ -1172,7 +1172,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- MergeSortOperator.class.getSimpleName());
+ TreeMergeSortOperator.class.getSimpleName());
List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
@@ -1237,7 +1237,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- TopKOperator.class.getSimpleName());
+ TreeTopKOperator.class.getSimpleName());
List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
context.setCachedDataTypes(dataTypes);
List<Operator> children = dealWithConsumeAllChildrenPipelineBreaker(node, context);
@@ -1251,7 +1251,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
sortItemList,
sortItemIndexList,
sortItemDataTypeList);
- return new TopKOperator(
+ return new TreeTopKOperator(
operatorContext,
children,
dataTypes,
@@ -2180,7 +2180,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- SortOperator.class.getSimpleName());
+ TreeSortOperator.class.getSimpleName());
List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
@@ -2208,7 +2208,7 @@ public class OperatorTreeGenerator extends PlanVisitor<Operator, LocalExecutionP
Operator child = node.getChild().accept(this, context);
- return new SortOperator(
+ return new TreeSortOperator(
operatorContext,
child,
dataTypes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
index b1ba6ded7f0..b81bf8cd264 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/TableOperatorGenerator.java
@@ -35,11 +35,11 @@ import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.CollectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.FilterAndProjectOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.LimitOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.MergeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.OffsetOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.StreamSortOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.TopKOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TableMergeSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TableSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TableStreamSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TableTopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaQueryScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSourceFactory;
import org.apache.iotdb.db.queryengine.execution.operator.sink.IdentitySinkOperator;
@@ -47,7 +47,6 @@ import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesSc
import org.apache.iotdb.db.queryengine.execution.operator.source.ExchangeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
import org.apache.iotdb.db.queryengine.execution.relational.ColumnTransformerBuilder;
-import org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils;
import org.apache.iotdb.db.queryengine.plan.analyze.TypeProvider;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
@@ -104,7 +103,6 @@ import static org.apache.iotdb.db.queryengine.common.DataNodeEndPoints.isSameNod
import static org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.MergeSortComparator.getComparatorForTable;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator.constructAlignedPath;
import static org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
-import static org.apache.iotdb.db.queryengine.plan.expression.leaf.TimestampOperand.TIMESTAMP_EXPRESSION_STRING;
import static org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager.getTSDataType;
/** This Visitor is responsible for transferring Table PlanNode Tree to Table Operator Tree. */
@@ -150,7 +148,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
sinkHandle.setMaxBytesCanReserve(context.getMaxBytesOneHandleCanReserve());
context.getDriverContext().setSink(sinkHandle);
- // List<Operator> children = dealWithConsumeChildrenOneByOneNode(node, context);
Operator child = node.getChildren().get(0).accept(this, context);
List<Operator> children = new ArrayList<>(1);
children.add(child);
@@ -211,7 +208,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
int measurementColumnCount = 0;
int idx = 0;
- boolean hasTimeColumn = false;
for (Symbol columnName : outputColumnNames) {
ColumnSchema schema =
requireNonNull(columnSchemaMap.get(columnName), columnName + " is null");
@@ -233,8 +229,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
columnSchemas.add(schema);
break;
case TIME:
- hasTimeColumn = true;
- // columnsIndexArray[i] = -1;
+ columnsIndexArray[idx++] = -1;
break;
default:
throw new IllegalArgumentException(
@@ -242,11 +237,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
}
}
- int[] newColumnsIndexArray = new int[outputColumnCount - 1];
- if (hasTimeColumn) {
- System.arraycopy(columnsIndexArray, 0, newColumnsIndexArray, 0, outputColumnCount - 1);
- }
-
SeriesScanOptions.Builder scanOptionsBuilder =
node.getTimePredicate()
.map(timePredicate -> getSeriesScanOptionsBuilder(context, timePredicate))
@@ -283,7 +273,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
operatorContext,
node.getPlanNodeId(),
columnSchemas,
- newColumnsIndexArray,
+ columnsIndexArray,
measurementColumnCount,
node.getDeviceEntries(),
node.getScanOrder(),
@@ -313,16 +303,10 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
int valueColumnIndex = 0;
for (Symbol columnName : childNode.getOutputSymbols()) {
- if (!TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(columnName.getName())) {
- outputMappings
- .computeIfAbsent(columnName, key -> new ArrayList<>())
- .add(new InputLocation(tsBlockIndex, valueColumnIndex));
- valueColumnIndex++;
- } else {
- outputMappings
- .computeIfAbsent(columnName, key -> new ArrayList<>())
- .add(new InputLocation(tsBlockIndex, -1));
- }
+ outputMappings
+ .computeIfAbsent(columnName, key -> new ArrayList<>())
+ .add(new InputLocation(tsBlockIndex, valueColumnIndex));
+ valueColumnIndex++;
}
tsBlockIndex++;
}
@@ -341,10 +325,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return scanOptionsBuilder;
}
- private boolean canPushIntoScan(Expression pushDownPredicate) {
- return pushDownPredicate == null || PredicateUtils.predicateCanPushIntoScan(pushDownPredicate);
- }
-
@Override
public Operator visitFilter(FilterNode node, LocalExecutionPlanContext context) {
TypeProvider typeProvider = context.getTypeProvider();
@@ -356,10 +336,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return constructFilterAndProjectOperator(
predicate,
inputOperator,
- node.getOutputSymbols().stream()
- .filter(symbol -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(symbol.getName()))
- .map(Symbol::toSymbolReference)
- .toArray(Expression[]::new),
+ node.getOutputSymbols().stream().map(Symbol::toSymbolReference).toArray(Expression[]::new),
inputDataTypes,
inputLocations,
node.getPlanNodeId(),
@@ -479,11 +456,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return constructFilterAndProjectOperator(
predicate,
inputOperator,
- node.getAssignments().getMap().entrySet().stream()
- .filter(
- entry -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(entry.getKey().getName()))
- .map(Map.Entry::getValue)
- .toArray(Expression[]::new),
+ node.getAssignments().getMap().values().toArray(new Expression[0]),
inputDataTypes,
inputLocations,
node.getPlanNodeId(),
@@ -495,7 +468,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return node.getChildren().stream()
.map(PlanNode::getOutputSymbols)
.flatMap(List::stream)
- .filter(s -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(s.getName()))
.map(s -> getTSDataType(typeProvider.getTableModelType(s)))
.collect(Collectors.toList());
}
@@ -551,10 +523,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
return constructFilterAndProjectOperator(
predicate,
inputOperator,
- node.getOutputSymbols().stream()
- .filter(e -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(e.getName()))
- .map(Symbol::toSymbolReference)
- .toArray(Expression[]::new),
+ node.getOutputSymbols().stream().map(Symbol::toSymbolReference).toArray(Expression[]::new),
inputDataTypes,
inputLocations,
node.getPlanNodeId(),
@@ -585,7 +554,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- MergeSortOperator.class.getSimpleName());
+ TableMergeSortOperator.class.getSimpleName());
List<Operator> children = new ArrayList<>(node.getChildren().size());
for (PlanNode child : node.getChildren()) {
children.add(this.process(child, context));
@@ -602,7 +571,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
sortItemDataTypeList,
context.getTypeProvider());
- return new MergeSortOperator(
+ return new TableMergeSortOperator(
operatorContext,
children,
dataTypes,
@@ -618,7 +587,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- SortOperator.class.getSimpleName());
+ TableSortOperator.class.getSimpleName());
List<TSDataType> dataTypes = getOutputColumnTypes(node, context.getTypeProvider());
int sortItemsCount = node.getOrderingScheme().getOrderBy().size();
@@ -644,7 +613,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Operator child = node.getChild().accept(this, context);
- return new SortOperator(
+ return new TableSortOperator(
operatorContext,
child,
dataTypes,
@@ -661,7 +630,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
.addOperatorContext(
context.getNextOperatorId(),
node.getPlanNodeId(),
- TopKOperator.class.getSimpleName());
+ TableTopKOperator.class.getSimpleName());
List<Operator> children = new ArrayList<>(node.getChildren().size());
for (PlanNode child : node.getChildren()) {
children.add(this.process(child, context));
@@ -677,7 +646,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
sortItemIndexList,
sortItemDataTypeList,
context.getTypeProvider());
- return new TopKOperator(
+ return new TableTopKOperator(
operatorContext,
children,
dataTypes,
@@ -689,7 +658,6 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
private List<TSDataType> getOutputColumnTypes(PlanNode node, TypeProvider typeProvider) {
return node.getOutputSymbols().stream()
- .filter(s -> !TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(s.getName()))
.map(s -> getTSDataType(typeProvider.getTableModelType(s)))
.collect(Collectors.toList());
}
@@ -703,27 +671,19 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Map<Symbol, Integer> columnIndex = new HashMap<>();
int index = 0;
for (Symbol symbol : outputSymbols) {
- if (TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(symbol.getName())) {
- continue;
- }
columnIndex.put(symbol, index++);
}
orderingScheme
.getOrderBy()
.forEach(
sortItem -> {
- if (TIMESTAMP_EXPRESSION_STRING.equalsIgnoreCase(sortItem.getName())) {
- sortItemIndexList.add(-1);
- sortItemDataTypeList.add(TSDataType.INT64);
- } else {
- Integer i = columnIndex.get(sortItem);
- if (i == null) {
- throw new IllegalStateException(
- "Sort Item %s is not included in children's output columns");
- }
- sortItemIndexList.add(i);
- sortItemDataTypeList.add(getTSDataType(typeProvider.getTableModelType(sortItem)));
+ Integer i = columnIndex.get(sortItem);
+ if (i == null) {
+ throw new IllegalStateException(
+ "Sort Item %s is not included in children's output columns");
}
+ sortItemIndexList.add(i);
+ sortItemDataTypeList.add(getTSDataType(typeProvider.getTableModelType(sortItem)));
});
}
@@ -761,7 +721,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Operator child = node.getChild().accept(this, context);
- return new StreamSortOperator(
+ return new TableStreamSortOperator(
operatorContext,
child,
dataTypes,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/SpecifiedInfoMergerFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/SpecifiedInfoMergerFactory.java
index 08e0ffd6b12..e9bbff7d2d2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/SpecifiedInfoMergerFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/statistics/SpecifiedInfoMergerFactory.java
@@ -43,8 +43,8 @@ public class SpecifiedInfoMergerFactory {
public static SpecifiedInfoMerger getMerger(String operatorType) {
switch (operatorType) {
- case "SortOperator":
- case "MergeSortOperator":
+ case "TreeSortOperator":
+ case "TreeMergeSortOperator":
case "FilterAndProjectOperator":
return LONG_MERGER;
default:
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputMultiReferenceLayer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputMultiReferenceLayer.java
index a419c71c535..99814831b96 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputMultiReferenceLayer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputMultiReferenceLayer.java
@@ -77,7 +77,7 @@ public class SingleInputMultiReferenceLayer extends IntermediateLayer {
return new LayerReader() {
private final SafetyPile safetyPile = safetyLine.addSafetyPile();
- private TimeColumn cachedTimes = null;
+ private Column cachedTimes = null;
private Column cachedValues = null;
private int cacheConsumed = 0;
@@ -242,10 +242,10 @@ public class SingleInputMultiReferenceLayer extends IntermediateLayer {
private long currentEndTime = Long.MAX_VALUE;
private final TVListForwardIterator beginIterator = tvList.constructIterator();
- private TimeColumn cachedBeginTimeColumn;
+ private Column cachedBeginTimeColumn;
private int cachedBeginConsumed;
- private TimeColumn cachedEndTimeColumn;
+ private Column cachedEndTimeColumn;
private int cachedEndConsumed;
@Override
@@ -269,11 +269,12 @@ public class SingleInputMultiReferenceLayer extends IntermediateLayer {
if (nextWindowTimeBegin == Long.MIN_VALUE) {
// display window begin should be set to the same as the min timestamp of the query
// result set
- nextWindowTimeBegin = cachedEndTimeColumn.getStartTime();
+ nextWindowTimeBegin = cachedEndTimeColumn.getLong(0);
}
hasAtLeastOneRow = tvList.size() != 0;
if (hasAtLeastOneRow) {
- currentEndTime = cachedEndTimeColumn.getEndTime();
+ currentEndTime =
+ cachedEndTimeColumn.getLong(cachedEndTimeColumn.getPositionCount() - 1);
}
isFirstIteration = false;
}
@@ -347,7 +348,8 @@ public class SingleInputMultiReferenceLayer extends IntermediateLayer {
}
if ((nextIndexEnd == nextIndexBegin)
- && nextWindowTimeEnd < cachedEndTimeColumn.getEndTime()) {
+ && nextWindowTimeEnd
+ < cachedEndTimeColumn.getLong(cachedEndTimeColumn.getPositionCount() - 1)) {
window.setEmptyWindow(nextWindowTimeBegin, nextWindowTimeEnd);
return YieldableState.YIELDABLE;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputSingleReferenceLayer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputSingleReferenceLayer.java
index fdb0779867d..dadcb8a2271 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputSingleReferenceLayer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/dag/intermediate/SingleInputSingleReferenceLayer.java
@@ -165,10 +165,10 @@ public class SingleInputSingleReferenceLayer extends IntermediateLayer {
private long currentEndTime = Long.MAX_VALUE;
private final TVListForwardIterator beginIterator = tvList.constructIterator();
- private TimeColumn cachedBeginTimeColumn;
+ private Column cachedBeginTimeColumn;
private int cachedBeginConsumed;
- private TimeColumn cachedEndTimeColumn;
+ private Column cachedEndTimeColumn;
private int cachedEndConsumed;
@Override
@@ -192,11 +192,12 @@ public class SingleInputSingleReferenceLayer extends IntermediateLayer {
if (nextWindowTimeBegin == Long.MIN_VALUE) {
// display window begin should be set to the same as the min timestamp of the query
// result set
- nextWindowTimeBegin = cachedEndTimeColumn.getStartTime();
+ nextWindowTimeBegin = cachedEndTimeColumn.getLong(0);
}
hasAtLeastOneRow = tvList.size() != 0;
if (hasAtLeastOneRow) {
- currentEndTime = cachedEndTimeColumn.getEndTime();
+ currentEndTime =
+ cachedEndTimeColumn.getLong(cachedEndTimeColumn.getPositionCount() - 1);
}
isFirstIteration = false;
}
@@ -270,7 +271,8 @@ public class SingleInputSingleReferenceLayer extends IntermediateLayer {
}
if ((nextIndexEnd == nextIndexBegin)
- && nextWindowTimeEnd < cachedEndTimeColumn.getEndTime()) {
+ && nextWindowTimeEnd
+ < cachedEndTimeColumn.getLong(cachedEndTimeColumn.getPositionCount() - 1)) {
window.setEmptyWindow(nextWindowTimeBegin, nextWindowTimeEnd);
return YieldableState.YIELDABLE;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/iterator/TVListForwardIterator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/iterator/TVListForwardIterator.java
index 134961afd45..cd36f417ba7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/iterator/TVListForwardIterator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/iterator/TVListForwardIterator.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.db.queryengine.transformation.datastructure.tv.ElasticSe
import org.apache.iotdb.db.queryengine.transformation.datastructure.tv.SerializableTVList;
import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import java.io.IOException;
@@ -54,7 +53,7 @@ public class TVListForwardIterator implements ListForwardIterator {
endPointIndex = tvList.getLastPointIndex(externalIndex, internalIndex);
}
- public TimeColumn currentTimes() throws IOException {
+ public Column currentTimes() throws IOException {
return tvList.getTimeColumn(externalIndex, internalIndex);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableBinaryTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableBinaryTVList.java
index fd6d111ef45..e596a3662af 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableBinaryTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableBinaryTVList.java
@@ -52,7 +52,7 @@ public class ElasticSerializableBinaryTVList extends ElasticSerializableTVList {
}
@Override
- public void putColumn(TimeColumn timeColumn, Column valueColumn) throws IOException {
+ public void putColumn(Column timeColumn, Column valueColumn) throws IOException {
super.putColumn(timeColumn, valueColumn);
// Update two memory control statistics
@@ -176,10 +176,10 @@ public class ElasticSerializableBinaryTVList extends ElasticSerializableTVList {
TVListForwardIterator iterator =
new TVListForwardIterator(this, externalColumnIndex, internalColumnIndex);
// Get and put split columns after eviction upper bound
- TimeColumn timeColumn = iterator.currentTimes();
+ Column timeColumn = iterator.currentTimes();
Column valueColumn = iterator.currentValues();
if (tvOffsetInColumns != 0) {
- timeColumn = (TimeColumn) timeColumn.subColumnCopy(tvOffsetInColumns);
+ timeColumn = timeColumn.subColumnCopy(tvOffsetInColumns);
valueColumn = valueColumn.subColumnCopy(tvOffsetInColumns);
}
newESTVList.putColumn(timeColumn, valueColumn);
@@ -199,7 +199,7 @@ public class ElasticSerializableBinaryTVList extends ElasticSerializableTVList {
private void copyColumnByIterator(ElasticSerializableTVList target, TVListForwardIterator source)
throws IOException {
- TimeColumn timeColumn = source.currentTimes();
+ Column timeColumn = source.currentTimes();
Column valueColumn = source.currentValues();
target.putColumn(timeColumn, valueColumn);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableTVList.java
index 324aa38fee3..f0d4e3dd174 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/ElasticSerializableTVList.java
@@ -25,7 +25,6 @@ import org.apache.iotdb.db.queryengine.transformation.datastructure.iterator.TVL
import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.utils.Binary;
import java.io.IOException;
@@ -163,7 +162,7 @@ public class ElasticSerializableTVList {
// endregion
// region batch data points methods
- public TimeColumn getTimeColumn(int externalIndex, int internalIndex) throws IOException {
+ public Column getTimeColumn(int externalIndex, int internalIndex) throws IOException {
return cache.get(externalIndex).getTimeColumn(internalIndex);
}
@@ -171,14 +170,14 @@ public class ElasticSerializableTVList {
return cache.get(externalIndex).getValueColumn(internalIndex);
}
- public void putColumn(TimeColumn timeColumn, Column valueColumn) throws IOException {
+ public void putColumn(Column timeColumn, Column valueColumn) throws IOException {
checkExpansion();
int begin = 0, end = 0;
int total = timeColumn.getPositionCount();
while (total > 0) {
int consumed;
- TimeColumn insertedTimeColumn;
+ Column insertedTimeColumn;
Column insertedValueColumn;
if (total + pointCount % internalTVListCapacity < internalTVListCapacity) {
consumed = total;
@@ -187,13 +186,13 @@ public class ElasticSerializableTVList {
insertedTimeColumn = timeColumn;
insertedValueColumn = valueColumn;
} else {
- insertedTimeColumn = (TimeColumn) timeColumn.getRegionCopy(begin, consumed);
+ insertedTimeColumn = timeColumn.getRegionCopy(begin, consumed);
insertedValueColumn = valueColumn.getRegionCopy(begin, consumed);
}
} else {
consumed = internalTVListCapacity - pointCount % internalTVListCapacity;
// Construct sub-regions
- insertedTimeColumn = (TimeColumn) timeColumn.getRegionCopy(begin, consumed);
+ insertedTimeColumn = timeColumn.getRegionCopy(begin, consumed);
insertedValueColumn = valueColumn.getRegionCopy(begin, consumed);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/SerializableTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/SerializableTVList.java
index 0c8030f01ee..d29926947fe 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/SerializableTVList.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/tv/SerializableTVList.java
@@ -26,7 +26,6 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.read.common.block.TsBlock;
-import org.apache.tsfile.read.common.block.column.TimeColumn;
import org.apache.tsfile.read.common.block.column.TsBlockSerde;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.PublicBAOS;
@@ -50,7 +49,7 @@ public class SerializableTVList implements SerializableList {
private List<Column> valueColumns;
- private List<TimeColumn> timeColumns;
+ private List<Column> timeColumns;
private final List<Integer> columnSizes;
@@ -129,7 +128,7 @@ public class SerializableTVList implements SerializableList {
long ret = 0;
int total = 0;
- for (TimeColumn column : timeColumns) {
+ for (Column column : timeColumns) {
int length = column.getPositionCount();
if (index < total + length) {
int offset = index - total;
@@ -271,7 +270,7 @@ public class SerializableTVList implements SerializableList {
// endregion
// region batch data method
- public TimeColumn getTimeColumn(int index) {
+ public Column getTimeColumn(int index) {
checkState(index < timeColumns.size());
return timeColumns.get(index);
}
@@ -281,7 +280,7 @@ public class SerializableTVList implements SerializableList {
return valueColumns.get(index);
}
- public void putColumns(TimeColumn timeColumn, Column valueColumn) {
+ public void putColumns(Column timeColumn, Column valueColumn) {
timeColumns.add(timeColumn);
valueColumns.add(valueColumn);
@@ -360,7 +359,7 @@ public class SerializableTVList implements SerializableList {
for (int i = 0; i < timeColumns.size(); i++) {
// Construct TsBlock
- TimeColumn timeColumn = timeColumns.get(i);
+ Column timeColumn = timeColumns.get(i);
Column valueColumn = valueColumns.get(i);
TsBlock tsBlock = new TsBlock(timeColumn, valueColumn);
@@ -386,7 +385,7 @@ public class SerializableTVList implements SerializableList {
TsBlock tsBlock = serde.deserialize(byteBuffer);
// Unpack columns
- TimeColumn timeColumn = tsBlock.getTimeColumn();
+ Column timeColumn = tsBlock.getTimeColumn();
Column valueColumn = tsBlock.getColumn(0);
timeColumns.add(timeColumn);
@@ -395,7 +394,7 @@ public class SerializableTVList implements SerializableList {
}
@TestOnly
- public List<TimeColumn> getTimeColumns() {
+ public List<Column> getTimeColumns() {
return timeColumns;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java
index 7c10fb2fb2b..1c6791c8598 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointCrossCompactionWriter.java
@@ -46,7 +46,7 @@ public class ReadPointCrossCompactionWriter extends AbstractCrossCompactionWrite
@Override
public void write(TsBlock tsBlock, int subTaskId) throws IOException {
- TimeColumn timestamps = tsBlock.getTimeColumn();
+ TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn();
Column[] columns = tsBlock.getValueColumns();
int batchSize = tsBlock.getPositionCount();
// Since a batch of data is written, the end time of the current aligned device may exceed the
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java
index d57ffe95d0e..dcba826f86e 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/execute/utils/writer/ReadPointInnerCompactionWriter.java
@@ -41,7 +41,7 @@ public class ReadPointInnerCompactionWriter extends AbstractInnerCompactionWrite
@Override
public void write(TsBlock tsBlock, int subTaskId) throws IOException {
- TimeColumn timestamps = tsBlock.getTimeColumn();
+ TimeColumn timestamps = (TimeColumn) tsBlock.getTimeColumn();
Column[] columns = tsBlock.getValueColumns();
int batchSize = tsBlock.getPositionCount();
AlignedChunkWriterImpl chunkWriter = (AlignedChunkWriterImpl) this.chunkWriters[subTaskId];
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQuerySortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQueryTreeSortOperatorTest.java
similarity index 99%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQuerySortOperatorTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQueryTreeSortOperatorTest.java
index 3b9ce7756c1..e7a187ce12f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQuerySortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/LastQueryTreeSortOperatorTest.java
@@ -67,9 +67,9 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-public class LastQuerySortOperatorTest {
+public class LastQueryTreeSortOperatorTest {
- private static final String SERIES_SCAN_OPERATOR_TEST_SG = "root.LastQuerySortOperatorTest";
+ private static final String SERIES_SCAN_OPERATOR_TEST_SG = "root.LastQueryTreeSortOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
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/MergeTreeSortOperatorTest.java
similarity index 92%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeSortOperatorTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/MergeTreeSortOperatorTest.java
index 99519e84b93..8f89b41b92a 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/MergeTreeSortOperatorTest.java
@@ -31,9 +31,9 @@ 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.TreeMergeSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.FullOuterTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
@@ -95,9 +95,9 @@ import static org.junit.Assert.fail;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.when;
-public class MergeSortOperatorTest {
+public class MergeTreeSortOperatorTest {
- private static final String MERGE_SORT_OPERATOR_TEST_SG = "root.MergeSortOperatorTest";
+ private static final String MERGE_SORT_OPERATOR_TEST_SG = "root.MergeTreeSortOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
@@ -144,7 +144,7 @@ public class MergeSortOperatorTest {
// ------------------------------------------------------------------------------------------------
// order by time - 1
// ------------------------------------------------------------------------------------------------
- // MergeSortOperator
+ // TreeMergeSortOperator
// ____________|_______________
// / | \
// SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
@@ -153,7 +153,8 @@ public class MergeSortOperatorTest {
// / \ / \
// SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
// ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TreeMergeSortOperator mergeSortOperatorTest(
+ Ordering timeOrdering, Ordering deviceOrdering) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
// Construct operator tree
@@ -188,7 +189,7 @@ public class MergeSortOperatorTest {
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 11, new PlanNodeId("11"), MergeSortOperator.class.getSimpleName());
+ 11, new PlanNodeId("11"), TreeMergeSortOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -343,8 +344,8 @@ public class MergeSortOperatorTest {
Arrays.asList(4, 5),
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TreeMergeSortOperator treeMergeSortOperator =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(10),
Arrays.asList(
singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
@@ -355,18 +356,20 @@ public class MergeSortOperatorTest {
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;
+ treeMergeSortOperator
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return treeMergeSortOperator;
}
@Test
public void testOrderByTime1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.ASC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -409,12 +412,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.ASC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ mergeSortOperatorTest(Ordering.ASC, Ordering.DESC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -457,12 +461,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -505,12 +510,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest(Ordering.DESC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(6, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -556,10 +562,10 @@ public class MergeSortOperatorTest {
// ------------------------------------------------------------------------------------------------
// [SSO]:SeriesScanOperator
// [SDO]:SingleDeviceOperator
- // MergeSortOperator
+ // TreeMergeSortOperator
// ___________|___________
// / \
- // MergeSortOperator MergeSortOperator
+ // TreeMergeSortOperator TreeMergeSortOperator
// / | / \
// [SDO] [SDO] [SDO] [SDO]
// | | | |
@@ -568,7 +574,8 @@ public class MergeSortOperatorTest {
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest2(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TreeMergeSortOperator mergeSortOperatorTest2(
+ Ordering timeOrdering, Ordering deviceOrdering) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
// Construct operator tree
@@ -611,11 +618,11 @@ 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"), TreeMergeSortOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 16, new PlanNodeId("16"), MergeSortOperator.class.getSimpleName());
+ 16, new PlanNodeId("16"), TreeMergeSortOperator.class.getSimpleName());
driverContext.addOperatorContext(
- 17, new PlanNodeId("17"), MergeSortOperator.class.getSimpleName());
+ 17, new PlanNodeId("17"), TreeMergeSortOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -822,8 +829,8 @@ public class MergeSortOperatorTest {
Arrays.asList(1, 2),
tsDataTypes);
- MergeSortOperator mergeSortOperator1 =
- new MergeSortOperator(
+ TreeMergeSortOperator treeMergeSortOperator1 =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(14),
Arrays.asList(singleDeviceViewOperator1, singleDeviceViewOperator2),
tsDataTypes,
@@ -833,9 +840,11 @@ public class MergeSortOperatorTest {
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(
+ treeMergeSortOperator1
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ TreeMergeSortOperator treeMergeSortOperator2 =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(15),
Arrays.asList(singleDeviceViewOperator3, singleDeviceViewOperator4),
tsDataTypes,
@@ -845,12 +854,14 @@ public class MergeSortOperatorTest {
new SortItem(OrderByKey.DEVICE, deviceOrdering)),
Arrays.asList(-1, 0),
Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
- mergeSortOperator2.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ treeMergeSortOperator2
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TreeMergeSortOperator treeMergeSortOperator =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(16),
- Arrays.asList(mergeSortOperator1, mergeSortOperator2),
+ Arrays.asList(treeMergeSortOperator1, treeMergeSortOperator2),
tsDataTypes,
MergeSortComparator.getComparator(
Arrays.asList(
@@ -858,18 +869,21 @@ public class MergeSortOperatorTest {
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;
+ treeMergeSortOperator
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return treeMergeSortOperator;
}
@Test
public void testOrderByTime1_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ mergeSortOperatorTest2(Ordering.ASC, Ordering.ASC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -908,13 +922,14 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime2_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.ASC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ mergeSortOperatorTest2(Ordering.ASC, Ordering.DESC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -953,12 +968,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime3_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -997,12 +1013,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByTime4_2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest2(Ordering.DESC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1043,7 +1060,7 @@ public class MergeSortOperatorTest {
// order by device
// ------------------------------------------------------------------------------------------------
// [SSO]:SeriesScanOperator
- // MergeSortOperator
+ // TreeMergeSortOperator
// ___________|___________
// / \
// DeviceViewOperator DeviceViewOperator
@@ -1053,7 +1070,8 @@ public class MergeSortOperatorTest {
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------------------
- public MergeSortOperator mergeSortOperatorTest3(Ordering timeOrdering, Ordering deviceOrdering) {
+ public TreeMergeSortOperator mergeSortOperatorTest3(
+ Ordering timeOrdering, Ordering deviceOrdering) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
// Construct operator tree
@@ -1092,7 +1110,7 @@ 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"), TreeMergeSortOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -1301,8 +1319,8 @@ public class MergeSortOperatorTest {
: Arrays.asList(timeJoinOperator3, timeJoinOperator2),
deviceColumnIndex,
tsDataTypes);
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TreeMergeSortOperator treeMergeSortOperator =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(12),
Arrays.asList(deviceViewOperator1, deviceViewOperator2),
tsDataTypes,
@@ -1312,18 +1330,21 @@ public class MergeSortOperatorTest {
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;
+ treeMergeSortOperator
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return treeMergeSortOperator;
}
@Test
public void testOrderByDevice1() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ mergeSortOperatorTest3(Ordering.ASC, Ordering.ASC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1371,12 +1392,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByDevice2() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.ASC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ mergeSortOperatorTest3(Ordering.ASC, Ordering.DESC);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock tsBlock = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1424,13 +1446,14 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByDevice3() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.ASC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1477,12 +1500,13 @@ public class MergeSortOperatorTest {
@Test
public void testOrderByDevice4() throws Exception {
- MergeSortOperator mergeSortOperator = mergeSortOperatorTest3(Ordering.DESC, Ordering.DESC);
+ TreeMergeSortOperator treeMergeSortOperator =
+ 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();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock tsBlock = treeMergeSortOperator.next();
if (tsBlock == null) continue;
assertEquals(3, tsBlock.getValueColumnCount());
count += tsBlock.getPositionCount();
@@ -1531,10 +1555,10 @@ public class MergeSortOperatorTest {
// order by Time, DataNodeID
// ------------------------------------------------------------------------------------------------
//
- // MergeSortOperator
+ // TreeMergeSortOperator
// ___________|__________
// / \
- // SortOperator SortOperator
+ // TreeSortOperator TreeSortOperator
// | |
// ShowQueriesOperator ShowQueriesOperator
// ------------------------------------------------------------------------------------------------
@@ -1559,11 +1583,11 @@ public class MergeSortOperatorTest {
PlanNodeId planNodeId1 = new PlanNodeId("1");
driverContext.addOperatorContext(1, planNodeId1, ShowQueriesOperator.class.getSimpleName());
PlanNodeId planNodeId2 = new PlanNodeId("2");
- driverContext.addOperatorContext(2, planNodeId2, SortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(2, planNodeId2, TreeSortOperator.class.getSimpleName());
PlanNodeId planNodeId3 = new PlanNodeId("3");
- driverContext.addOperatorContext(3, planNodeId3, SortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(3, planNodeId3, TreeSortOperator.class.getSimpleName());
PlanNodeId planNodeId4 = new PlanNodeId("4");
- driverContext.addOperatorContext(4, planNodeId4, MergeSortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(4, planNodeId4, TreeMergeSortOperator.class.getSimpleName());
List<OperatorContext> operatorContexts = driverContext.getOperatorContexts();
List<TSDataType> dataTypes = DatasetHeaderFactory.getShowQueriesHeader().getRespDataTypes();
@@ -1594,19 +1618,19 @@ public class MergeSortOperatorTest {
new ShowQueriesOperator(operatorContexts.get(0), planNodeId0, coordinator1);
ShowQueriesOperator showQueriesOperator2 =
new ShowQueriesOperator(operatorContexts.get(1), planNodeId1, coordinator2);
- SortOperator sortOperator1 =
- new SortOperator(
+ TreeSortOperator treeSortOperator1 =
+ new TreeSortOperator(
operatorContexts.get(2), showQueriesOperator1, dataTypes, "", comparator);
- SortOperator sortOperator2 =
- new SortOperator(
+ TreeSortOperator treeSortOperator2 =
+ new TreeSortOperator(
operatorContexts.get(3), showQueriesOperator2, dataTypes, "", comparator);
Operator root =
- new MergeSortOperator(
+ new TreeMergeSortOperator(
operatorContexts.get(4),
new ArrayList<Operator>() {
{
- add(sortOperator1);
- add(sortOperator2);
+ add(treeSortOperator1);
+ add(treeSortOperator2);
}
},
dataTypes,
@@ -1749,10 +1773,10 @@ public class MergeSortOperatorTest {
createFragmentInstanceContext(instanceId, stateMachine);
DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0);
PlanNodeId planNodeId1 = new PlanNodeId("1");
- driverContext.addOperatorContext(1, planNodeId1, MergeSortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(1, planNodeId1, TreeMergeSortOperator.class.getSimpleName());
- MergeSortOperator mergeSortOperator =
- new MergeSortOperator(
+ TreeMergeSortOperator treeMergeSortOperator =
+ new TreeMergeSortOperator(
driverContext.getOperatorContexts().get(0),
Arrays.asList(childOperator1, childOperator2),
Collections.singletonList(TSDataType.INT64),
@@ -1760,11 +1784,13 @@ public class MergeSortOperatorTest {
Collections.singletonList(new SortItem(OrderByKey.TIME, Ordering.ASC)),
Collections.singletonList(-1),
Collections.singletonList(TSDataType.INT64)));
- mergeSortOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ treeMergeSortOperator
+ .getOperatorContext()
+ .setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
int index = 0;
- while (mergeSortOperator.isBlocked().isDone() && mergeSortOperator.hasNext()) {
- TsBlock result = mergeSortOperator.next();
+ while (treeMergeSortOperator.isBlocked().isDone() && treeMergeSortOperator.hasNext()) {
+ TsBlock result = treeMergeSortOperator.next();
for (int i = 0; i < result.getPositionCount(); i++) {
long time = result.getTimeByIndex(i);
assertEquals(time, ans[index++]);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
index 1b9a6d29f3f..e22a2df6a0a 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorMemoryTest.java
@@ -45,7 +45,7 @@ import org.apache.iotdb.db.queryengine.execution.operator.process.LinearFillOper
import org.apache.iotdb.db.queryengine.execution.operator.process.OffsetOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.RawDataAggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.SlidingWindowAggregationOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.process.SortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.IFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.fill.linear.LinearFill;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.FullOuterTimeJoinOperator;
@@ -512,8 +512,8 @@ public class OperatorMemoryTest {
Mockito.when(child.calculateMaxReturnSize()).thenReturn(1024L);
Mockito.when(child.calculateRetainedSizeAfterCallingNext()).thenReturn(512L);
- SortOperator sortOperator =
- new SortOperator(
+ TreeSortOperator treeSortOperator =
+ new TreeSortOperator(
Mockito.mock(OperatorContext.class),
child,
Collections.singletonList(TSDataType.INT32),
@@ -522,13 +522,13 @@ public class OperatorMemoryTest {
assertEquals(
2048 + 512 + IoTDBDescriptor.getInstance().getConfig().getSortBufferSize(),
- sortOperator.calculateMaxPeekMemory());
+ treeSortOperator.calculateMaxPeekMemory());
assertEquals(
TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes(),
- sortOperator.calculateMaxReturnSize());
+ treeSortOperator.calculateMaxReturnSize());
assertEquals(
512 + IoTDBDescriptor.getInstance().getConfig().getSortBufferSize(),
- sortOperator.calculateRetainedSizeAfterCallingNext());
+ treeSortOperator.calculateRetainedSizeAfterCallingNext());
}
@Test
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/SortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeSortOperatorTest.java
similarity index 94%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/SortOperatorTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeSortOperatorTest.java
index 99acfd45366..94d47c14f00 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/SortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeSortOperatorTest.java
@@ -29,7 +29,7 @@ import org.apache.iotdb.db.queryengine.common.QueryId;
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.SortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeSortOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.FullOuterTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
@@ -71,9 +71,9 @@ import static org.apache.iotdb.db.utils.EnvironmentUtils.cleanDir;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-public class SortOperatorTest {
+public class TreeSortOperatorTest {
- private static final String SORT_OPERATOR_TEST_SG = "root.SortOperatorTest";
+ private static final String SORT_OPERATOR_TEST_SG = "root.TreeSortOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
@@ -108,7 +108,7 @@ public class SortOperatorTest {
// ------------------------------------------------------------------------------------------------
// sortOperatorTest
// ------------------------------------------------------------------------------------------------
- // SortOperator
+ // TreeSortOperator
// |
// TimeJoinOperator
// _____________________|______________________________
@@ -119,7 +119,7 @@ public class SortOperatorTest {
// ------------------------------------------------------------------------------------------------
public Operator genSortOperator(Ordering timeOrdering, boolean getSortOperator) {
ExecutorService instanceNotificationExecutor =
- IoTDBThreadPoolFactory.newFixedThreadPool(1, "sortOperator-test-instance-notification");
+ IoTDBThreadPoolFactory.newFixedThreadPool(1, "treeSortOperator-test-instance-notification");
// Construct operator tree
QueryId queryId = new QueryId("stub_query");
@@ -136,7 +136,8 @@ public class SortOperatorTest {
driverContext.addOperatorContext(2, planNodeId2, SeriesScanOperator.class.getSimpleName());
driverContext.addOperatorContext(
3, new PlanNodeId("3"), FullOuterTimeJoinOperator.class.getSimpleName());
- driverContext.addOperatorContext(4, new PlanNodeId("4"), SortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 4, new PlanNodeId("4"), TreeSortOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -212,10 +213,11 @@ public class SortOperatorTest {
+ File.separator
+ operatorContext.getDriverContext().getPipelineId()
+ File.separator;
- SortOperator sortOperator =
- new SortOperator(operatorContext, timeJoinOperator1, tsDataTypes, filePrefix, comparator);
- sortOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- return sortOperator;
+ TreeSortOperator treeSortOperator =
+ new TreeSortOperator(
+ operatorContext, timeJoinOperator1, tsDataTypes, filePrefix, comparator);
+ treeSortOperator.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
+ return treeSortOperator;
}
long getValue(long expectedTime) {
@@ -235,7 +237,7 @@ public class SortOperatorTest {
public void sortOperatorSpillingTest() throws Exception {
IoTDBDescriptor.getInstance().getConfig().setSortBufferSize(5000);
long sortBufferSize = IoTDBDescriptor.getInstance().getConfig().getSortBufferSize();
- try (SortOperator root = (SortOperator) genSortOperator(Ordering.ASC, true)) {
+ try (TreeSortOperator root = (TreeSortOperator) genSortOperator(Ordering.ASC, true)) {
int lastValue = -1;
int count = 0;
while (root.isBlocked().isDone() && root.hasNext()) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeTopKOperatorTest.java
similarity index 96%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeTopKOperatorTest.java
index f8a4f104d96..59bf0292d7e 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TopKOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/TreeTopKOperatorTest.java
@@ -31,9 +31,9 @@ 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.TopKOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeMergeSortOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.process.TreeTopKOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.FullOuterTimeJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.AscTimeComparator;
import org.apache.iotdb.db.queryengine.execution.operator.process.join.merge.DescTimeComparator;
@@ -87,12 +87,9 @@ import static org.junit.Assert.fail;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.when;
-/**
- * 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";
+/** The original data and structure of this test is same as {@link MergeTreeSortOperatorTest}. */
+public class TreeTopKOperatorTest {
+ private static final String TOP_K_OPERATOR_TEST_SG = "root.TreeTopKOperatorTest";
private final List<String> deviceIds = new ArrayList<>();
private final List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
@@ -146,7 +143,7 @@ public class TopKOperatorTest {
// ----------------------------------------------------------------------------------------------
// order by time - 1
// ----------------------------------------------------------------------------------------------
- // TopKOperator
+ // TreeTopKOperator
// ____________|_______________
// / | \
// SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
@@ -155,7 +152,7 @@ public class TopKOperatorTest {
// / \ / \
// SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
// ----------------------------------------------------------------------------------------------
- public TopKOperator topKOperatorTest(
+ public TreeTopKOperator topKOperatorTest(
Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
@@ -189,7 +186,8 @@ public class TopKOperatorTest {
9, new PlanNodeId("9"), FullOuterTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
- driverContext.addOperatorContext(11, new PlanNodeId("11"), TopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 11, new PlanNodeId("11"), TreeTopKOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -334,8 +332,8 @@ public class TopKOperatorTest {
Arrays.asList(4, 5),
tsDataTypes);
- TopKOperator topKOperator =
- new TopKOperator(
+ TreeTopKOperator topKOperator =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(10),
Arrays.asList(
singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
@@ -354,7 +352,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByTime1() throws Exception {
- TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -404,7 +402,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByTime2() throws Exception {
- TopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.DESC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -452,7 +450,7 @@ public class TopKOperatorTest {
assertEquals(limitValue, count);
}
- public TopKOperator topKOperatorTest2(
+ public TreeTopKOperator topKOperatorTest2(
Ordering timeOrdering, Ordering deviceOrdering, int limitValue) throws IllegalPathException {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
@@ -495,9 +493,12 @@ public class TopKOperatorTest {
13, new PlanNodeId("13"), SingleDeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
14, new PlanNodeId("14"), SingleDeviceViewOperator.class.getSimpleName());
- driverContext.addOperatorContext(15, new PlanNodeId("15"), TopKOperator.class.getSimpleName());
- driverContext.addOperatorContext(16, new PlanNodeId("16"), TopKOperator.class.getSimpleName());
- driverContext.addOperatorContext(17, new PlanNodeId("17"), TopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 15, new PlanNodeId("15"), TreeTopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 16, new PlanNodeId("16"), TreeTopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 17, new PlanNodeId("17"), TreeTopKOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -702,8 +703,8 @@ public class TopKOperatorTest {
Arrays.asList(1, 2),
tsDataTypes);
- TopKOperator topKOperator1 =
- new TopKOperator(
+ TreeTopKOperator topKOperator1 =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(14),
Arrays.asList(singleDeviceViewOperator1, singleDeviceViewOperator2),
tsDataTypes,
@@ -716,8 +717,8 @@ public class TopKOperatorTest {
limitValue,
true);
topKOperator1.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TopKOperator topKOperator2 =
- new TopKOperator(
+ TreeTopKOperator topKOperator2 =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(15),
Arrays.asList(singleDeviceViewOperator3, singleDeviceViewOperator4),
tsDataTypes,
@@ -731,8 +732,8 @@ public class TopKOperatorTest {
true);
topKOperator2.getOperatorContext().setMaxRunTime(new Duration(500, TimeUnit.MILLISECONDS));
- TopKOperator topKOperator =
- new TopKOperator(
+ TreeTopKOperator topKOperator =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(16),
Arrays.asList(topKOperator1, topKOperator2),
tsDataTypes,
@@ -750,7 +751,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByTime1_2() throws Exception {
- TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.ASC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -796,7 +797,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByTime2_2() throws Exception {
- TopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.DESC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest2(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -845,7 +846,7 @@ public class TopKOperatorTest {
// order by device
// -----------------------------------------------------------------------------------
// [SSO]:SeriesScanOperator
- // TopKOperator
+ // TreeTopKOperator
// ___________|___________
// / \
// DeviceViewOperator DeviceViewOperator
@@ -855,7 +856,7 @@ public class TopKOperatorTest {
// / \ / \ / \
// [SSO] [SSO] [SSO] [SSO] [SSO] [SSO]
// ------------------------------------------------------------------------------------
- public TopKOperator topKOperatorTest3(
+ public TreeTopKOperator topKOperatorTest3(
Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
@@ -893,7 +894,8 @@ public class TopKOperatorTest {
11, new PlanNodeId("11"), DeviceViewOperator.class.getSimpleName());
driverContext.addOperatorContext(
12, new PlanNodeId("12"), DeviceViewOperator.class.getSimpleName());
- driverContext.addOperatorContext(13, new PlanNodeId("13"), TopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 13, new PlanNodeId("13"), TreeTopKOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -1084,8 +1086,8 @@ public class TopKOperatorTest {
: Arrays.asList(timeJoinOperator3, timeJoinOperator2),
deviceColumnIndex,
tsDataTypes);
- TopKOperator topKOperator =
- new TopKOperator(
+ TreeTopKOperator topKOperator =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(12),
Arrays.asList(deviceViewOperator1, deviceViewOperator2),
tsDataTypes,
@@ -1103,7 +1105,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByDevice1() throws Exception {
- TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.ASC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.ASC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -1158,7 +1160,7 @@ public class TopKOperatorTest {
@Test
public void testOrderByDevice2() throws Exception {
- TopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.DESC, limitValue);
+ TreeTopKOperator topKOperator = topKOperatorTest3(Ordering.ASC, Ordering.DESC, limitValue);
long lastTime = -1;
int checkDevice = 0;
int count = 0;
@@ -1311,10 +1313,10 @@ public class TopKOperatorTest {
createFragmentInstanceContext(instanceId, stateMachine);
DriverContext driverContext = new DriverContext(fragmentInstanceContext, 0);
PlanNodeId planNodeId1 = new PlanNodeId("1");
- driverContext.addOperatorContext(1, planNodeId1, MergeSortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(1, planNodeId1, TreeMergeSortOperator.class.getSimpleName());
- TopKOperator topKOperator =
- new TopKOperator(
+ TreeTopKOperator topKOperator =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(0),
Arrays.asList(childOperator1, childOperator2),
Collections.singletonList(TSDataType.INT64),
@@ -1343,7 +1345,7 @@ public class TopKOperatorTest {
// ----------------------------------------------------------------------------------------------
// order by time - all result of SeriesScan is empty
// ----------------------------------------------------------------------------------------------
- // TopKOperator
+ // TreeTopKOperator
// ____________|_______________
// / | \
// SingleDeviceViewOperator SingleDeviceViewOperator SingleDeviceViewOperator
@@ -1352,7 +1354,7 @@ public class TopKOperatorTest {
// / \ / \
// SeriesScanOperator SeriesScanOperator SeriesScanOperator SeriesScanOperator
// ----------------------------------------------------------------------------------------------
- public TopKOperator emptyTopKOperatorTest(
+ public TreeTopKOperator emptyTopKOperatorTest(
Ordering timeOrdering, Ordering deviceOrdering, int limitValue) {
ExecutorService instanceNotificationExecutor =
IoTDBThreadPoolFactory.newFixedThreadPool(1, "test-instance-notification");
@@ -1386,7 +1388,8 @@ public class TopKOperatorTest {
9, new PlanNodeId("9"), FullOuterTimeJoinOperator.class.getSimpleName());
driverContext.addOperatorContext(
10, new PlanNodeId("10"), SingleDeviceViewOperator.class.getSimpleName());
- driverContext.addOperatorContext(11, new PlanNodeId("11"), TopKOperator.class.getSimpleName());
+ driverContext.addOperatorContext(
+ 11, new PlanNodeId("11"), TreeTopKOperator.class.getSimpleName());
NonAlignedFullPath measurementPath1 =
new NonAlignedFullPath(
@@ -1536,8 +1539,8 @@ public class TopKOperatorTest {
Arrays.asList(4, 5),
tsDataTypes);
- TopKOperator topKOperator =
- new TopKOperator(
+ TreeTopKOperator topKOperator =
+ new TreeTopKOperator(
driverContext.getOperatorContexts().get(10),
Arrays.asList(
singleDeviceViewOperator1, singleDeviceViewOperator2, singleDeviceViewOperator3),
@@ -1556,7 +1559,7 @@ public class TopKOperatorTest {
@Test
public void testEmptyTopKOperator() throws Exception {
- TopKOperator topKOperator = emptyTopKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
+ TreeTopKOperator topKOperator = emptyTopKOperatorTest(Ordering.ASC, Ordering.ASC, limitValue);
while (topKOperator.isBlocked().isDone() && topKOperator.hasNext()) {
TsBlock tsBlock = topKOperator.next();
if (tsBlock == null || tsBlock.isEmpty()) {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperatorTest.java
similarity index 93%
rename from iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperatorTest.java
rename to iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperatorTest.java
index f273a183604..39148577aa6 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/StreamSortOperatorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/process/TableStreamSortOperatorTest.java
@@ -58,7 +58,7 @@ import static org.apache.iotdb.db.utils.EnvironmentUtils.cleanDir;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
-public class StreamSortOperatorTest {
+public class TableStreamSortOperatorTest {
private static final String sortTmpPrefixPath =
"target" + File.separator + "sort" + File.separator + "tmp";
@@ -151,7 +151,7 @@ public class StreamSortOperatorTest {
@Test
public void allInMemoryTest() {
- try (StreamSortOperator streamSortOperator = genStreamSortOperator(1000)) {
+ try (TableStreamSortOperator streamSortOperator = genStreamSortOperator(1000)) {
int count = 0;
ListenableFuture<?> listenableFuture = streamSortOperator.isBlocked();
listenableFuture.get();
@@ -192,12 +192,12 @@ public class StreamSortOperatorTest {
public void allInMemoryTes2() {
int maxTsBlockLineNumber = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
TSFileDescriptor.getInstance().getConfig().setMaxTsBlockLineNumber(2);
- try (StreamSortOperator streamSortOperator = genStreamSortOperator(2)) {
+ try (TableStreamSortOperator tableStreamSortOperator = genStreamSortOperator(2)) {
int count = 0;
- ListenableFuture<?> listenableFuture = streamSortOperator.isBlocked();
+ ListenableFuture<?> listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
- while (!streamSortOperator.isFinished() && streamSortOperator.hasNext()) {
- TsBlock tsBlock = streamSortOperator.next();
+ while (!tableStreamSortOperator.isFinished() && tableStreamSortOperator.hasNext()) {
+ TsBlock tsBlock = tableStreamSortOperator.next();
if (tsBlock != null && !tsBlock.isEmpty()) {
assertEquals(2, tsBlock.getPositionCount());
for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) {
@@ -220,7 +220,7 @@ public class StreamSortOperatorTest {
}
}
}
- listenableFuture = streamSortOperator.isBlocked();
+ listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
}
assertEquals(timeArray.length, count);
@@ -240,12 +240,12 @@ public class StreamSortOperatorTest {
TSFileDescriptor.getInstance().getConfig().getMaxTsBlockSizeInBytes();
IoTDBDescriptor.getInstance().getConfig().setSortBufferSize(500);
TSFileDescriptor.getInstance().getConfig().setMaxTsBlockSizeInBytes(50);
- try (StreamSortOperator streamSortOperator = genStreamSortOperator(1000)) {
+ try (TableStreamSortOperator tableStreamSortOperator = genStreamSortOperator(1000)) {
int count = 0;
- ListenableFuture<?> listenableFuture = streamSortOperator.isBlocked();
+ ListenableFuture<?> listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
- while (!streamSortOperator.isFinished() && streamSortOperator.hasNext()) {
- TsBlock tsBlock = streamSortOperator.next();
+ while (!tableStreamSortOperator.isFinished() && tableStreamSortOperator.hasNext()) {
+ TsBlock tsBlock = tableStreamSortOperator.next();
if (tsBlock != null && !tsBlock.isEmpty()) {
for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) {
assertEquals(timeArray[count], tsBlock.getTimeByIndex(i));
@@ -267,7 +267,7 @@ public class StreamSortOperatorTest {
}
}
}
- listenableFuture = streamSortOperator.isBlocked();
+ listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
}
assertEquals(timeArray.length, count);
@@ -289,13 +289,13 @@ public class StreamSortOperatorTest {
IoTDBDescriptor.getInstance().getConfig().setSortBufferSize(500);
TSFileDescriptor.getInstance().getConfig().setMaxTsBlockSizeInBytes(50);
TSFileDescriptor.getInstance().getConfig().setMaxTsBlockLineNumber(2);
- try (StreamSortOperator streamSortOperator = genStreamSortOperator(2)) {
+ try (TableStreamSortOperator tableStreamSortOperator = genStreamSortOperator(2)) {
int count = 0;
- ListenableFuture<?> listenableFuture = streamSortOperator.isBlocked();
+ ListenableFuture<?> listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
- while (!streamSortOperator.isFinished() && streamSortOperator.hasNext()) {
- TsBlock tsBlock = streamSortOperator.next();
+ while (!tableStreamSortOperator.isFinished() && tableStreamSortOperator.hasNext()) {
+ TsBlock tsBlock = tableStreamSortOperator.next();
if (tsBlock != null && !tsBlock.isEmpty()) {
assertEquals(2, tsBlock.getPositionCount());
for (int i = 0, size = tsBlock.getPositionCount(); i < size; i++, count++) {
@@ -318,7 +318,7 @@ public class StreamSortOperatorTest {
}
}
}
- listenableFuture = streamSortOperator.isBlocked();
+ listenableFuture = tableStreamSortOperator.isBlocked();
listenableFuture.get();
}
assertEquals(timeArray.length, count);
@@ -332,7 +332,7 @@ public class StreamSortOperatorTest {
}
}
- private StreamSortOperator genStreamSortOperator(int maxLinesToOutput) {
+ private TableStreamSortOperator genStreamSortOperator(int maxLinesToOutput) {
// child output
// Time, city, deviceId, s1
// 1 null d1 9
@@ -395,7 +395,7 @@ public class StreamSortOperatorTest {
PlanNodeId planNodeId1 = new PlanNodeId("1");
driverContext.addOperatorContext(1, planNodeId1, TableScanOperator.class.getSimpleName());
PlanNodeId planNodeId2 = new PlanNodeId("2");
- driverContext.addOperatorContext(2, planNodeId2, StreamSortOperator.class.getSimpleName());
+ driverContext.addOperatorContext(2, planNodeId2, TableStreamSortOperator.class.getSimpleName());
Operator childOperator =
new Operator() {
@@ -579,7 +579,7 @@ public class StreamSortOperatorTest {
sortItemIndexList.subList(0, 2),
sortItemDataTypeList.subList(0, 2));
- return new StreamSortOperator(
+ return new TableStreamSortOperator(
operatorContext,
childOperator,
Arrays.asList(TSDataType.TEXT, TSDataType.TEXT, TSDataType.INT32),
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/transformation/datastructure/SerializableTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/transformation/datastructure/SerializableTVListTest.java
index 0ee4dfa84a6..bf3f566be7d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/transformation/datastructure/SerializableTVListTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/transformation/datastructure/SerializableTVListTest.java
@@ -424,7 +424,7 @@ public class SerializableTVListTest extends SerializableListTest {
private static class ForwardIterator {
int index;
int offset;
- List<TimeColumn> timeColumns;
+ List<Column> timeColumns;
List<Column> valueColumns;
ForwardIterator(SerializableTVList tvList) {