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 2025/01/02 13:30:43 UTC
(iotdb) branch master updated: Enable tree data to be queried by a table view [PART-1]
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 256737f8e25 Enable tree data to be queried by a table view [PART-1]
256737f8e25 is described below
commit 256737f8e2558b1ac6ae6decb37a5602d098a619
Author: Jackie Tien <ja...@gmail.com>
AuthorDate: Thu Jan 2 21:30:36 2025 +0800
Enable tree data to be queried by a table view [PART-1]
---
.../relational/AbstractAggTableScanOperator.java | 241 +++++------
.../AbstractDefaultAggTableScanOperator.java | 109 +++++
...perator.java => AbstractTableScanOperator.java} | 94 +++--
.../relational/DefaultAggTableScanOperator.java | 58 +--
.../relational/LastQueryAggTableScanOperator.java | 66 +--
.../source/relational/TableScanOperator.java | 332 +--------------
...eAlignedDeviceViewAggregationScanOperator.java} | 28 +-
.../TreeAlignedDeviceViewScanOperator.java} | 27 +-
.../plan/planner/TableOperatorGenerator.java | 462 ++++++++++++++-------
.../plan/planner/plan/node/PlanGraphPrinter.java | 24 +-
.../plan/planner/plan/node/PlanNode.java | 2 +
.../plan/planner/plan/node/PlanNodeType.java | 12 +
.../plan/planner/plan/node/PlanVisitor.java | 20 +
.../plan/planner/plan/node/source/SourceNode.java | 1 +
.../relational/metadata/AlignedDeviceEntry.java | 56 +++
.../plan/relational/metadata/DeviceEntry.java | 42 +-
.../metadata/NonAlignedAlignedDeviceEntry.java | 56 +++
.../metadata/TreeDeviceViewSchema.java} | 23 +-
.../metadata/fetcher/TableDeviceSchemaFetcher.java | 5 +-
.../plan/relational/planner/RelationPlanner.java | 20 +
.../distribute/TableDistributedPlanGenerator.java | 238 +++++++++--
.../iterative/rule/PruneTableScanColumns.java | 52 ++-
.../planner/node/AggregationTableScanNode.java | 387 ++++++++---------
.../node/AggregationTreeDeviceViewScanNode.java | 205 +++++++++
.../planner/node/DeviceTableScanNode.java | 197 ++++-----
.../node/InformationSchemaTableScanNode.java | 105 +----
.../relational/planner/node/TableScanNode.java | 118 +++++-
.../node/TreeAlignedDeviceViewScanNode.java | 126 ++++++
.../planner/node/TreeDeviceViewScanNode.java | 208 ++++++++++
.../node/TreeNonAlignedDeviceViewScanNode.java | 126 ++++++
.../PushAggregationIntoTableScan.java | 4 +
.../optimizations/PushPredicateIntoTableScan.java | 14 +-
.../optimizations/UnaliasSymbolReferences.java | 40 +-
.../relational/planner/optimizations/Util.java | 73 +++-
.../analyzer/MockTableModelDataPartition.java | 10 +-
.../plan/relational/analyzer/TSBSMetadata.java | 23 +-
.../plan/relational/analyzer/TestMatadata.java | 138 ++++--
.../plan/relational/analyzer/TestPlanBuilder.java | 6 +-
.../plan/relational/analyzer/TreeViewTest.java | 198 +++++++++
.../relational/planner/assertions/PlanAssert.java | 2 +-
.../planner/assertions/PlanMatchPattern.java | 72 ++++
41 files changed, 2692 insertions(+), 1328 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggTableScanOperator.java
index 04c54394f03..ac4b611b2fd 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggTableScanOperator.java
@@ -21,7 +21,6 @@ package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
import org.apache.iotdb.commons.path.AlignedFullPath;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
-import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITableTimeRangeIterator;
import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
@@ -31,6 +30,7 @@ import org.apache.iotdb.db.queryengine.execution.operator.window.IWindow;
import org.apache.iotdb.db.queryengine.execution.operator.window.TimeWindow;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
@@ -52,7 +52,6 @@ import org.apache.tsfile.read.common.block.column.BinaryColumn;
import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.Pair;
-import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.io.IOException;
@@ -61,7 +60,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.apache.iotdb.db.queryengine.execution.operator.AggregationUtil.satisfiedTimeRange;
@@ -73,9 +71,6 @@ import static org.apache.tsfile.read.common.block.TsBlockUtil.skipPointsOutOfTim
public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOperator {
- private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(AbstractAggTableScanOperator.class);
-
private boolean finished = false;
private TsBlock inputTsBlock;
@@ -116,53 +111,39 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
private boolean allAggregatorsHasFinalResult = false;
- public AbstractAggTableScanOperator(
- PlanNodeId sourceId,
- OperatorContext context,
- List<ColumnSchema> aggColumnSchemas,
- int[] aggColumnsIndexArray,
- List<DeviceEntry> deviceEntries,
- int deviceCount,
- SeriesScanOptions seriesScanOptions,
- List<String> measurementColumnNames,
- Set<String> allSensors,
- List<IMeasurementSchema> measurementSchemas,
- List<TableAggregator> tableAggregators,
- List<ColumnSchema> groupingKeySchemas,
- int[] groupingKeyIndex,
- ITableTimeRangeIterator tableTimeRangeIterator,
- boolean ascending,
- boolean canUseStatistics,
- List<Integer> aggregatorInputChannels) {
-
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.canUseStatistics = canUseStatistics;
- this.tableAggregators = tableAggregators;
- this.groupingKeySchemas = groupingKeySchemas;
- this.groupingKeyIndex = groupingKeyIndex;
- this.groupingKeySize = groupingKeySchemas == null ? 0 : groupingKeySchemas.size();
- this.aggColumnSchemas = aggColumnSchemas;
- this.aggColumnsIndexArray = aggColumnsIndexArray;
- this.deviceEntries = deviceEntries;
- this.deviceCount = deviceCount;
+ public AbstractAggTableScanOperator(AbstractAggTableScanOperatorParameter parameter) {
+
+ this.sourceId = parameter.sourceId;
+ this.operatorContext = parameter.context;
+ this.canUseStatistics = parameter.canUseStatistics;
+ this.tableAggregators = parameter.tableAggregators;
+ this.groupingKeySchemas = parameter.groupingKeySchemas;
+ this.groupingKeyIndex = parameter.groupingKeyIndex;
+ this.groupingKeySize =
+ parameter.groupingKeySchemas == null ? 0 : parameter.groupingKeySchemas.size();
+ this.aggColumnSchemas = parameter.aggColumnSchemas;
+ this.aggColumnsIndexArray = parameter.aggColumnsIndexArray;
+ this.deviceEntries = parameter.deviceEntries;
+ this.deviceCount = parameter.deviceCount;
this.operatorContext.recordSpecifiedInfo(DEVICE_NUMBER, Integer.toString(this.deviceCount));
- this.ascending = ascending;
- this.scanOrder = ascending ? Ordering.ASC : Ordering.DESC;
- this.seriesScanOptions = seriesScanOptions;
- this.measurementColumnNames = measurementColumnNames;
+ this.ascending = parameter.ascending;
+ this.scanOrder = parameter.ascending ? Ordering.ASC : Ordering.DESC;
+ this.seriesScanOptions = parameter.seriesScanOptions;
+ this.measurementColumnNames = parameter.measurementColumnNames;
this.measurementCount = measurementColumnNames.size();
this.cachedRawDataSize =
(1L + this.measurementCount)
* TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
- this.allSensors = allSensors;
- this.measurementSchemas = measurementSchemas;
+ this.allSensors = parameter.allSensors;
+ this.measurementSchemas = parameter.measurementSchemas;
this.measurementColumnTSDataTypes =
- measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
+ parameter.measurementSchemas.stream()
+ .map(IMeasurementSchema::getType)
+ .collect(Collectors.toList());
this.currentDeviceIndex = 0;
this.operatorContext.recordSpecifiedInfo(CURRENT_DEVICE_INDEX_STRING, Integer.toString(0));
- this.aggregatorInputChannels = aggregatorInputChannels;
- this.timeIterator = tableTimeRangeIterator;
+ this.aggregatorInputChannels = parameter.aggregatorInputChannels;
+ this.timeIterator = parameter.tableTimeRangeIterator;
this.dateBinSize =
timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.DATE_BIN_TIME_ITERATOR
? 1
@@ -179,56 +160,6 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
return finished;
}
- @Override
- public boolean hasNext() throws Exception {
- if (retainedTsBlock != null) {
- return true;
- }
-
- return timeIterator.hasCachedTimeRange() || timeIterator.hasNextTimeRange();
- }
-
- @Override
- public TsBlock next() throws Exception {
- if (retainedTsBlock != null) {
- return getResultFromRetainedTsBlock();
- }
-
- // optimize for sql: select count(*) from (select count(s1), sum(s1) from table)
- if (tableAggregators.isEmpty()
- && timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR
- && resultTsBlockBuilder.getValueColumnBuilders().length == 0) {
- resultTsBlockBuilder.reset();
- currentDeviceIndex = deviceCount;
- timeIterator.setFinished();
- Column[] valueColumns = new Column[0];
- return new TsBlock(1, new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, 1), valueColumns);
- }
-
- // start stopwatch, reset leftRuntimeOfOneNextCall
- long start = System.nanoTime();
- leftRuntimeOfOneNextCall = 1000 * operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
- long maxRuntime = leftRuntimeOfOneNextCall;
-
- while (System.nanoTime() - start < maxRuntime
- && (timeIterator.hasCachedTimeRange() || timeIterator.hasNextTimeRange())
- && !resultTsBlockBuilder.isFull()) {
-
- // calculate aggregation result on current time window
- // return true if current time window is calc finished
- if (calculateAggregationResultForCurrentTimeRange()) {
- timeIterator.resetCurTimeRange();
- }
- }
-
- if (resultTsBlockBuilder.isEmpty()) {
- return null;
- }
-
- buildResultTsBlock();
- return checkTsBlockSizeAndGetResult();
- }
-
protected abstract void updateResultTsBlock();
protected void buildResultTsBlock() {
@@ -244,7 +175,7 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
if (this.deviceEntries.isEmpty() || this.deviceEntries.get(this.currentDeviceIndex) == null) {
// for device which is not exist
- deviceEntry = new DeviceEntry(new StringArrayDeviceID(""), Collections.emptyList());
+ deviceEntry = new AlignedDeviceEntry(new StringArrayDeviceID(""), Collections.emptyList());
} else {
deviceEntry = this.deviceEntries.get(this.currentDeviceIndex);
}
@@ -422,12 +353,9 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
case TIME:
return inputRegion.getTimeColumn();
case TAG:
- // TODO avoid create deviceStatics multi times; count, sum can use time statistics
String id =
- (String)
- deviceEntries
- .get(currentDeviceIndex)
- .getNthSegment(aggColumnsIndexArray[columnIdx] + 1);
+ getNthIdColumnValue(
+ deviceEntries.get(currentDeviceIndex), aggColumnsIndexArray[columnIdx]);
return getIdOrAttrColumn(
inputRegion.getTimeColumn().getPositionCount(),
id == null ? null : new Binary(id, TSFileConfig.STRING_CHARSET));
@@ -493,12 +421,9 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
case TIME:
return timeStatistics;
case TAG:
- // TODO avoid create deviceStatics multi times; count, sum can use time statistics
String id =
- (String)
- deviceEntries
- .get(currentDeviceIndex)
- .getNthSegment(aggColumnsIndexArray[columnIdx] + 1);
+ getNthIdColumnValue(
+ deviceEntries.get(currentDeviceIndex), aggColumnsIndexArray[columnIdx]);
return getStatistics(
timeStatistics, id == null ? null : new Binary(id, TSFileConfig.STRING_CHARSET));
case ATTRIBUTE:
@@ -723,7 +648,7 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
ColumnBuilder[] columnBuilders = resultTsBlockBuilder.getValueColumnBuilders();
for (int i = 0; i < groupingKeySize; i++) {
if (TsTableColumnCategory.TAG == groupingKeySchemas.get(i).getColumnCategory()) {
- String id = (String) deviceEntries.get(deviceIndex).getNthSegment(groupingKeyIndex[i] + 1);
+ String id = getNthIdColumnValue(deviceEntries.get(deviceIndex), groupingKeyIndex[i]);
if (id == null) {
columnBuilders[i].appendNull();
} else {
@@ -839,19 +764,101 @@ public abstract class AbstractAggTableScanOperator extends AbstractDataSourceOpe
: 0;
}
- @Override
- public long ramBytesUsed() {
- return INSTANCE_SIZE
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(seriesScanUtil)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(sourceId)
- + (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
- + RamUsageEstimator.sizeOfCollection(deviceEntries);
- }
-
@Override
public void close() throws Exception {
super.close();
tableAggregators.forEach(TableAggregator::close);
}
+
+ abstract String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex);
+
+ public static class AbstractAggTableScanOperatorParameter {
+ private final String timeColumnName;
+ protected final PlanNodeId sourceId;
+ protected final OperatorContext context;
+ protected final List<ColumnSchema> aggColumnSchemas;
+ protected final int[] aggColumnsIndexArray;
+ protected final SeriesScanOptions seriesScanOptions;
+ protected final List<String> measurementColumnNames;
+ protected final Set<String> allSensors;
+ protected final List<IMeasurementSchema> measurementSchemas;
+ protected final List<TableAggregator> tableAggregators;
+ protected final List<ColumnSchema> groupingKeySchemas;
+ protected final int[] groupingKeyIndex;
+ protected final ITableTimeRangeIterator tableTimeRangeIterator;
+ protected final boolean ascending;
+ protected final boolean canUseStatistics;
+ protected final List<Integer> aggregatorInputChannels;
+
+ protected List<DeviceEntry> deviceEntries;
+ protected int deviceCount;
+
+ public AbstractAggTableScanOperatorParameter(
+ PlanNodeId sourceId,
+ OperatorContext context,
+ List<ColumnSchema> aggColumnSchemas,
+ int[] aggColumnsIndexArray,
+ List<DeviceEntry> deviceEntries,
+ int deviceCount,
+ SeriesScanOptions seriesScanOptions,
+ List<String> measurementColumnNames,
+ Set<String> allSensors,
+ List<IMeasurementSchema> measurementSchemas,
+ List<TableAggregator> tableAggregators,
+ List<ColumnSchema> groupingKeySchemas,
+ int[] groupingKeyIndex,
+ ITableTimeRangeIterator tableTimeRangeIterator,
+ boolean ascending,
+ boolean canUseStatistics,
+ List<Integer> aggregatorInputChannels,
+ String timeColumnName) {
+ this.sourceId = sourceId;
+ this.context = context;
+ this.aggColumnSchemas = aggColumnSchemas;
+ this.aggColumnsIndexArray = aggColumnsIndexArray;
+ this.deviceEntries = deviceEntries;
+ this.deviceCount = deviceCount;
+ this.seriesScanOptions = seriesScanOptions;
+ this.measurementColumnNames = measurementColumnNames;
+ this.allSensors = allSensors;
+ this.measurementSchemas = measurementSchemas;
+ this.tableAggregators = tableAggregators;
+ this.groupingKeySchemas = groupingKeySchemas;
+ this.groupingKeyIndex = groupingKeyIndex;
+ this.tableTimeRangeIterator = tableTimeRangeIterator;
+ this.ascending = ascending;
+ this.canUseStatistics = canUseStatistics;
+ this.aggregatorInputChannels = aggregatorInputChannels;
+ this.timeColumnName = timeColumnName;
+ }
+
+ public List<TableAggregator> getTableAggregators() {
+ return tableAggregators;
+ }
+
+ public SeriesScanOptions getSeriesScanOptions() {
+ return seriesScanOptions;
+ }
+
+ public Set<String> getAllSensors() {
+ return allSensors;
+ }
+
+ public List<String> getMeasurementColumnNames() {
+ return measurementColumnNames;
+ }
+
+ public List<IMeasurementSchema> getMeasurementSchemas() {
+ return measurementSchemas;
+ }
+
+ public String getTimeColumnName() {
+ return timeColumnName;
+ }
+
+ public void setDeviceEntries(List<DeviceEntry> deviceEntries) {
+ this.deviceEntries = deviceEntries;
+ this.deviceCount = deviceEntries.size();
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractDefaultAggTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractDefaultAggTableScanOperator.java
new file mode 100644
index 00000000000..f88aa22aabc
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractDefaultAggTableScanOperator.java
@@ -0,0 +1,109 @@
+/*
+ * 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.source.relational;
+
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
+import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITableTimeRangeIterator;
+
+import org.apache.tsfile.block.column.Column;
+import org.apache.tsfile.read.common.block.TsBlock;
+import org.apache.tsfile.read.common.block.column.RunLengthEncodedColumn;
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator.TIME_COLUMN_TEMPLATE;
+
+public abstract class AbstractDefaultAggTableScanOperator extends AbstractAggTableScanOperator {
+
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(AbstractDefaultAggTableScanOperator.class);
+
+ public AbstractDefaultAggTableScanOperator(AbstractAggTableScanOperatorParameter parameter) {
+ super(parameter);
+ }
+
+ @Override
+ public boolean hasNext() throws Exception {
+ if (retainedTsBlock != null) {
+ return true;
+ }
+
+ return timeIterator.hasCachedTimeRange() || timeIterator.hasNextTimeRange();
+ }
+
+ @Override
+ public TsBlock next() throws Exception {
+ if (retainedTsBlock != null) {
+ return getResultFromRetainedTsBlock();
+ }
+
+ // optimize for sql: select count(*) from (select count(s1), sum(s1) from table)
+ if (tableAggregators.isEmpty()
+ && timeIterator.getType() == ITableTimeRangeIterator.TimeIteratorType.SINGLE_TIME_ITERATOR
+ && resultTsBlockBuilder.getValueColumnBuilders().length == 0) {
+ resultTsBlockBuilder.reset();
+ currentDeviceIndex = deviceCount;
+ timeIterator.setFinished();
+ Column[] valueColumns = new Column[0];
+ return new TsBlock(1, new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, 1), valueColumns);
+ }
+
+ // start stopwatch, reset leftRuntimeOfOneNextCall
+ long start = System.nanoTime();
+ leftRuntimeOfOneNextCall = 1000 * operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
+ long maxRuntime = leftRuntimeOfOneNextCall;
+
+ while (System.nanoTime() - start < maxRuntime
+ && (timeIterator.hasCachedTimeRange() || timeIterator.hasNextTimeRange())
+ && !resultTsBlockBuilder.isFull()) {
+
+ // calculate aggregation result on current time window
+ // return true if current time window is calc finished
+ if (calculateAggregationResultForCurrentTimeRange()) {
+ timeIterator.resetCurTimeRange();
+ }
+ }
+
+ if (resultTsBlockBuilder.isEmpty()) {
+ return null;
+ }
+
+ buildResultTsBlock();
+ return checkTsBlockSizeAndGetResult();
+ }
+
+ @Override
+ protected void updateResultTsBlock() {
+ appendAggregationResult();
+ // after appendAggregationResult invoked, aggregators must be cleared
+ resetTableAggregators();
+ }
+
+ @Override
+ public long ramBytesUsed() {
+ return INSTANCE_SIZE
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(seriesScanUtil)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
+ + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(sourceId)
+ + (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
+ + RamUsageEstimator.sizeOfCollection(deviceEntries);
+ }
+}
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/AbstractTableScanOperator.java
similarity index 80%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TableScanOperator.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractTableScanOperator.java
index 6abf101ab60..d8f7a5f3359 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/AbstractTableScanOperator.java
@@ -57,7 +57,7 @@ import static org.apache.iotdb.db.queryengine.execution.operator.source.AlignedS
import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanGraphPrinter.DEVICE_NUMBER;
import static org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager.getTSDataType;
-public class TableScanOperator extends AbstractSeriesScanOperator {
+public abstract class AbstractTableScanOperator extends AbstractSeriesScanOperator {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(TableScanOperator.class);
@@ -95,41 +95,33 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
private int currentDeviceIndex;
- public TableScanOperator(
- OperatorContext context,
- PlanNodeId sourceId,
- List<ColumnSchema> columnSchemas,
- int[] columnsIndexArray,
- List<DeviceEntry> deviceEntries,
- Ordering scanOrder,
- SeriesScanOptions seriesScanOptions,
- List<String> measurementColumnNames,
- Set<String> allSensors,
- List<IMeasurementSchema> measurementSchemas,
- int maxTsBlockLineNum) {
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.operatorContext.recordSpecifiedInfo(DEVICE_NUMBER, Integer.toString(deviceEntries.size()));
- this.columnSchemas = columnSchemas;
- this.columnsIndexArray = columnsIndexArray;
- this.deviceEntries = deviceEntries;
- this.deviceCount = deviceEntries.size();
- this.scanOrder = scanOrder;
- this.seriesScanOptions = seriesScanOptions;
- this.measurementColumnNames = measurementColumnNames;
- this.allSensors = allSensors;
- this.measurementSchemas = measurementSchemas;
+ public AbstractTableScanOperator(AbstractTableScanOperatorParameter parameter) {
+ this.sourceId = parameter.sourceId;
+ this.operatorContext = parameter.context;
+ this.operatorContext.recordSpecifiedInfo(
+ DEVICE_NUMBER, Integer.toString(parameter.deviceEntries.size()));
+ this.columnSchemas = parameter.columnSchemas;
+ this.columnsIndexArray = parameter.columnsIndexArray;
+ this.deviceEntries = parameter.deviceEntries;
+ this.deviceCount = parameter.deviceEntries.size();
+ this.scanOrder = parameter.scanOrder;
+ this.seriesScanOptions = parameter.seriesScanOptions;
+ this.measurementColumnNames = parameter.measurementColumnNames;
+ this.allSensors = parameter.allSensors;
+ this.measurementSchemas = parameter.measurementSchemas;
this.measurementColumnTSDataTypes =
- measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
+ parameter.measurementSchemas.stream()
+ .map(IMeasurementSchema::getType)
+ .collect(Collectors.toList());
this.currentDeviceIndex = 0;
this.operatorContext.recordSpecifiedInfo(CURRENT_DEVICE_INDEX_STRING, Integer.toString(0));
this.maxReturnSize =
Math.min(
maxReturnSize,
- (1L + columnsIndexArray.length)
+ (1L + parameter.columnsIndexArray.length)
* TSFileDescriptor.getInstance().getConfig().getPageSizeInByte());
- this.maxTsBlockLineNum = maxTsBlockLineNum;
+ this.maxTsBlockLineNum = parameter.maxTsBlockLineNum;
constructAlignedSeriesScanUtil();
}
@@ -214,9 +206,8 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
for (int i = 0; i < columnsIndexArray.length; i++) {
switch (columnSchemas.get(i).getColumnCategory()) {
case TAG:
- // +1 for skip the table name segment
- String idColumnValue =
- ((String) currentDeviceEntry.getNthSegment(columnsIndexArray[i] + 1));
+ String idColumnValue = getNthIdColumnValue(currentDeviceEntry, columnsIndexArray[i]);
+
valueColumns[i] =
getIdOrAttributeValueColumn(
idColumnValue == null
@@ -247,6 +238,8 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
valueColumns);
}
+ abstract String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex);
+
private RunLengthEncodedColumn getIdOrAttributeValueColumn(Binary value, int positionCount) {
if (value == null) {
return new RunLengthEncodedColumn(
@@ -351,4 +344,43 @@ public class TableScanOperator extends AbstractSeriesScanOperator {
+ (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
+ RamUsageEstimator.sizeOfCollection(deviceEntries);
}
+
+ public static class AbstractTableScanOperatorParameter {
+ public final OperatorContext context;
+ public final PlanNodeId sourceId;
+ public final List<ColumnSchema> columnSchemas;
+ public final int[] columnsIndexArray;
+ public final List<DeviceEntry> deviceEntries;
+ public final Ordering scanOrder;
+ public final SeriesScanOptions seriesScanOptions;
+ public final List<String> measurementColumnNames;
+ public final Set<String> allSensors;
+ public final List<IMeasurementSchema> measurementSchemas;
+ public final int maxTsBlockLineNum;
+
+ public AbstractTableScanOperatorParameter(
+ Set<String> allSensors,
+ OperatorContext context,
+ PlanNodeId sourceId,
+ List<ColumnSchema> columnSchemas,
+ int[] columnsIndexArray,
+ List<DeviceEntry> deviceEntries,
+ Ordering scanOrder,
+ SeriesScanOptions seriesScanOptions,
+ List<String> measurementColumnNames,
+ List<IMeasurementSchema> measurementSchemas,
+ int maxTsBlockLineNum) {
+ this.allSensors = allSensors;
+ this.context = context;
+ this.sourceId = sourceId;
+ this.columnSchemas = columnSchemas;
+ this.columnsIndexArray = columnsIndexArray;
+ this.deviceEntries = deviceEntries;
+ this.scanOrder = scanOrder;
+ this.seriesScanOptions = seriesScanOptions;
+ this.measurementColumnNames = measurementColumnNames;
+ this.measurementSchemas = measurementSchemas;
+ this.maxTsBlockLineNum = maxTsBlockLineNum;
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DefaultAggTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DefaultAggTableScanOperator.java
index a49446500d5..3ebd7bed0b3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DefaultAggTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/DefaultAggTableScanOperator.java
@@ -19,63 +19,17 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITableTimeRangeIterator;
-import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
-import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
-import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
+public class DefaultAggTableScanOperator extends AbstractDefaultAggTableScanOperator {
-import java.util.List;
-import java.util.Set;
-
-public class DefaultAggTableScanOperator extends AbstractAggTableScanOperator {
-
- public DefaultAggTableScanOperator(
- PlanNodeId sourceId,
- OperatorContext context,
- List<ColumnSchema> aggColumnSchemas,
- int[] aggColumnsIndexArray,
- List<DeviceEntry> deviceEntries,
- int deviceCount,
- SeriesScanOptions seriesScanOptions,
- List<String> measurementColumnNames,
- Set<String> allSensors,
- List<IMeasurementSchema> measurementSchemas,
- List<TableAggregator> tableAggregators,
- List<ColumnSchema> groupingKeySchemas,
- int[] groupingKeyIndex,
- ITableTimeRangeIterator tableTimeRangeIterator,
- boolean ascending,
- boolean canUseStatistics,
- List<Integer> aggregatorInputChannels) {
- super(
- sourceId,
- context,
- aggColumnSchemas,
- aggColumnsIndexArray,
- deviceEntries,
- deviceCount,
- seriesScanOptions,
- measurementColumnNames,
- allSensors,
- measurementSchemas,
- tableAggregators,
- groupingKeySchemas,
- groupingKeyIndex,
- tableTimeRangeIterator,
- ascending,
- canUseStatistics,
- aggregatorInputChannels);
+ public DefaultAggTableScanOperator(AbstractAggTableScanOperatorParameter parameter) {
+ super(parameter);
}
@Override
- protected void updateResultTsBlock() {
- appendAggregationResult();
- // after appendAggregationResult invoked, aggregators must be cleared
- resetTableAggregators();
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ // +1 for skipping the table name segment
+ return ((String) deviceEntry.getNthSegment(idColumnIndex + 1));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
index a1ae289f95c..b74fb4bdf54 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
@@ -21,15 +21,11 @@ package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
-import org.apache.iotdb.db.queryengine.execution.aggregation.timerangeiterator.ITableTimeRangeIterator;
-import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
import org.apache.iotdb.db.queryengine.execution.operator.process.last.LastQueryUtil;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.LastAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.LastByDescAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.LastDescAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.TableAggregator;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
@@ -44,12 +40,10 @@ import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.utils.TsPrimitiveType;
import org.apache.tsfile.write.UnSupportedDataTypeException;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
import java.util.ArrayList;
import java.util.List;
import java.util.OptionalLong;
-import java.util.Set;
import java.util.concurrent.TimeUnit;
import static org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.Utils.serializeTimeValue;
@@ -67,7 +61,6 @@ public class LastQueryAggTableScanOperator extends AbstractAggTableScanOperator
private static final TableDeviceSchemaCache TABLE_DEVICE_SCHEMA_CACHE =
TableDeviceSchemaCache.getInstance();
- private boolean finished = false;
private final String dbName;
private int outputDeviceIndex;
private DeviceEntry currentDeviceEntry;
@@ -84,58 +77,27 @@ public class LastQueryAggTableScanOperator extends AbstractAggTableScanOperator
private int lastTimeAggregationIdx = -1;
public LastQueryAggTableScanOperator(
- PlanNodeId sourceId,
- OperatorContext context,
- List<ColumnSchema> aggColumnSchemas,
- int[] aggColumnsIndexArray,
- List<DeviceEntry> unCachedDeviceEntries,
+ AbstractAggTableScanOperatorParameter parameter,
List<DeviceEntry> cachedDeviceEntries,
- SeriesScanOptions seriesScanOptions,
- List<String> measurementColumnNames,
- Set<String> allSensors,
- List<IMeasurementSchema> measurementSchemas,
- List<TableAggregator> tableAggregators,
- List<ColumnSchema> groupingKeySchemas,
- int[] groupingKeyIndex,
- ITableTimeRangeIterator tableTimeRangeIterator,
- boolean ascending,
- boolean canUseStatistics,
- List<Integer> aggregatorInputChannels,
QualifiedObjectName qualifiedObjectName,
List<Integer> hitCachesIndexes,
List<Pair<OptionalLong, TsPrimitiveType[]>> hitCachedResults) {
- super(
- sourceId,
- context,
- aggColumnSchemas,
- aggColumnsIndexArray,
- unCachedDeviceEntries,
- unCachedDeviceEntries.size(),
- seriesScanOptions,
- measurementColumnNames,
- allSensors,
- measurementSchemas,
- tableAggregators,
- groupingKeySchemas,
- groupingKeyIndex,
- tableTimeRangeIterator,
- ascending,
- canUseStatistics,
- aggregatorInputChannels);
+ super(parameter);
// notice that: deviceEntries store all unCachedDeviceEntries
- this.allDeviceCount = cachedDeviceEntries.size() + unCachedDeviceEntries.size();
+ this.allDeviceCount = cachedDeviceEntries.size() + parameter.deviceCount;
this.cachedDeviceEntries = cachedDeviceEntries;
- this.needUpdateCache = LastQueryUtil.needUpdateCache(seriesScanOptions.getGlobalTimeFilter());
+ this.needUpdateCache =
+ LastQueryUtil.needUpdateCache(parameter.seriesScanOptions.getGlobalTimeFilter());
this.needUpdateNullEntry =
- LastQueryUtil.needUpdateNullEntry(seriesScanOptions.getGlobalTimeFilter());
+ LastQueryUtil.needUpdateNullEntry(parameter.seriesScanOptions.getGlobalTimeFilter());
this.hitCachesIndexes = hitCachesIndexes;
this.hitCachedResults = hitCachedResults;
this.dbName = qualifiedObjectName.getDatabaseName();
- for (int i = 0; i < tableAggregators.size(); i++) {
- if (tableAggregators.get(i).getAccumulator() instanceof LastAccumulator) {
+ for (int i = 0; i < parameter.tableAggregators.size(); i++) {
+ if (parameter.tableAggregators.get(i).getAccumulator() instanceof LastAccumulator) {
lastTimeAggregationIdx = i;
}
}
@@ -202,10 +164,8 @@ public class LastQueryAggTableScanOperator extends AbstractAggTableScanOperator
switch (category) {
case TAG:
String id =
- (String)
- cachedDeviceEntries
- .get(currentHitCacheIndex)
- .getNthSegment(aggColumnsIndexArray[columnIdx] + 1);
+ getNthIdColumnValue(
+ cachedDeviceEntries.get(currentHitCacheIndex), aggColumnsIndexArray[columnIdx]);
if (id == null) {
if (aggregator.getStep().isOutputPartial()) {
columnBuilder.writeBinary(
@@ -418,6 +378,12 @@ public class LastQueryAggTableScanOperator extends AbstractAggTableScanOperator
resetTableAggregators();
}
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ // +1 for skipping the table name segment
+ return ((String) deviceEntry.getNthSegment(idColumnIndex + 1));
+ }
+
private TsPrimitiveType cloneTsPrimitiveType(TsPrimitiveType originalValue) {
switch (originalValue.getDataType()) {
case BOOLEAN:
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 6abf101ab60..11f10ef3dae 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
@@ -19,336 +19,16 @@
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.commons.path.AlignedFullPath;
-import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
-import org.apache.iotdb.db.queryengine.execution.operator.OperatorContext;
-import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractSeriesScanOperator;
-import org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanUtil;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
-import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
-import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
-import org.apache.iotdb.db.storageengine.dataregion.read.IQueryDataSource;
-import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
-import org.apache.tsfile.block.column.Column;
-import org.apache.tsfile.common.conf.TSFileConfig;
-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.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;
-import org.apache.tsfile.write.schema.IMeasurementSchema;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-
-import static org.apache.iotdb.db.queryengine.execution.operator.source.AlignedSeriesScanOperator.appendDataIntoBuilder;
-import static org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanGraphPrinter.DEVICE_NUMBER;
-import static org.apache.iotdb.db.queryengine.plan.relational.type.InternalTypeManager.getTSDataType;
-
-public class TableScanOperator extends AbstractSeriesScanOperator {
- private static final long INSTANCE_SIZE =
- RamUsageEstimator.shallowSizeOfInstance(TableScanOperator.class);
-
- public static final String CURRENT_DEVICE_INDEX_STRING = "CurrentDeviceIndex";
-
- public static final LongColumn TIME_COLUMN_TEMPLATE =
- new LongColumn(1, Optional.empty(), new long[] {0});
-
- private final List<ColumnSchema> columnSchemas;
-
- private final int[] columnsIndexArray;
-
- private final List<DeviceEntry> deviceEntries;
-
- private final int deviceCount;
-
- private final Ordering scanOrder;
- private final SeriesScanOptions seriesScanOptions;
-
- private final List<String> measurementColumnNames;
-
- private final Set<String> allSensors;
-
- private final List<IMeasurementSchema> measurementSchemas;
-
- private final List<TSDataType> measurementColumnTSDataTypes;
-
- private TsBlockBuilder measurementDataBuilder;
-
- private final int maxTsBlockLineNum;
-
- private TsBlock measurementDataBlock;
-
- private QueryDataSource queryDataSource;
-
- private int currentDeviceIndex;
-
- public TableScanOperator(
- OperatorContext context,
- PlanNodeId sourceId,
- List<ColumnSchema> columnSchemas,
- int[] columnsIndexArray,
- List<DeviceEntry> deviceEntries,
- Ordering scanOrder,
- SeriesScanOptions seriesScanOptions,
- List<String> measurementColumnNames,
- Set<String> allSensors,
- List<IMeasurementSchema> measurementSchemas,
- int maxTsBlockLineNum) {
- this.sourceId = sourceId;
- this.operatorContext = context;
- this.operatorContext.recordSpecifiedInfo(DEVICE_NUMBER, Integer.toString(deviceEntries.size()));
- this.columnSchemas = columnSchemas;
- this.columnsIndexArray = columnsIndexArray;
- this.deviceEntries = deviceEntries;
- this.deviceCount = deviceEntries.size();
- this.scanOrder = scanOrder;
- this.seriesScanOptions = seriesScanOptions;
- this.measurementColumnNames = measurementColumnNames;
- this.allSensors = allSensors;
- this.measurementSchemas = measurementSchemas;
- this.measurementColumnTSDataTypes =
- measurementSchemas.stream().map(IMeasurementSchema::getType).collect(Collectors.toList());
- this.currentDeviceIndex = 0;
- this.operatorContext.recordSpecifiedInfo(CURRENT_DEVICE_INDEX_STRING, Integer.toString(0));
-
- this.maxReturnSize =
- Math.min(
- maxReturnSize,
- (1L + columnsIndexArray.length)
- * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte());
- this.maxTsBlockLineNum = maxTsBlockLineNum;
-
- constructAlignedSeriesScanUtil();
- }
-
- @Override
- public TsBlock next() throws Exception {
- if (retainedTsBlock != null) {
- return getResultFromRetainedTsBlock();
- }
-
- try {
-
- // start stopwatch
- long maxRuntime = operatorContext.getMaxRunTime().roundTo(TimeUnit.NANOSECONDS);
- long start = System.nanoTime();
-
- boolean currentDeviceNoMoreData = false;
-
- // here use do-while to promise doing this at least once
- do {
- /*
- * 1. consume page data firstly
- * 2. consume chunk data secondly
- * 3. consume next file finally
- */
- if (!readPageData() && !readChunkData() && !readFileData()) {
- currentDeviceNoMoreData = true;
- break;
- }
-
- } while (System.nanoTime() - start < maxRuntime
- && !measurementDataBuilder.isFull()
- && measurementDataBlock == null);
-
- // current device' data is consumed up
- if (measurementDataBuilder.isEmpty()
- && measurementDataBlock == null
- && currentDeviceNoMoreData) {
- currentDeviceIndex++;
- prepareForNextDevice();
- }
-
- } catch (IOException e) {
- throw new RuntimeException("Error happened while scanning the file", e);
- }
-
- // get all measurement column data and time column data
- if (!measurementDataBuilder.isEmpty()) {
- measurementDataBlock = measurementDataBuilder.build();
- measurementDataBuilder.reset();
- }
-
- // append id column and attribute column
- if (!isEmpty(measurementDataBlock)) {
- constructResultTsBlock();
- } else {
- return null;
- }
- measurementDataBlock = null;
- return checkTsBlockSizeAndGetResult();
- }
-
- @Override
- protected void appendToBuilder(TsBlock tsBlock) {
- if (measurementDataBuilder.isEmpty()
- && tsBlock.getPositionCount() >= measurementDataBuilder.getMaxTsBlockLineNumber()) {
- measurementDataBlock = tsBlock;
- return;
- }
- appendDataIntoBuilder(tsBlock, measurementDataBuilder);
- }
-
- @Override
- protected void buildResult(TsBlock tsBlock) {
- throw new UnsupportedOperationException();
- }
-
- private void constructResultTsBlock() {
- int positionCount = measurementDataBlock.getPositionCount();
- DeviceEntry currentDeviceEntry = deviceEntries.get(currentDeviceIndex);
- Column[] valueColumns = new Column[columnsIndexArray.length];
- for (int i = 0; i < columnsIndexArray.length; i++) {
- switch (columnSchemas.get(i).getColumnCategory()) {
- case TAG:
- // +1 for skip the table name segment
- String idColumnValue =
- ((String) currentDeviceEntry.getNthSegment(columnsIndexArray[i] + 1));
- valueColumns[i] =
- getIdOrAttributeValueColumn(
- idColumnValue == null
- ? null
- : new Binary(idColumnValue, TSFileConfig.STRING_CHARSET),
- positionCount);
- break;
- case ATTRIBUTE:
- Binary attributeColumnValue =
- currentDeviceEntry.getAttributeColumnValues().get(columnsIndexArray[i]);
- valueColumns[i] = getIdOrAttributeValueColumn(attributeColumnValue, positionCount);
- break;
- case FIELD:
- valueColumns[i] = measurementDataBlock.getColumn(columnsIndexArray[i]);
- break;
- case TIME:
- valueColumns[i] = measurementDataBlock.getTimeColumn();
- break;
- default:
- throw new IllegalArgumentException(
- "Unexpected column category: " + columnSchemas.get(i).getColumnCategory());
- }
- }
- this.resultTsBlock =
- new TsBlock(
- positionCount,
- new RunLengthEncodedColumn(TIME_COLUMN_TEMPLATE, positionCount),
- valueColumns);
- }
-
- private RunLengthEncodedColumn getIdOrAttributeValueColumn(Binary value, int positionCount) {
- if (value == null) {
- return new RunLengthEncodedColumn(
- new BinaryColumn(1, Optional.of(new boolean[] {true}), new Binary[] {null}),
- positionCount);
- } else {
- return new RunLengthEncodedColumn(
- new BinaryColumn(1, Optional.empty(), new Binary[] {value}), positionCount);
- }
- }
-
- @Override
- public boolean hasNext() throws Exception {
- return !isFinished();
- }
-
- @Override
- public boolean isFinished() throws Exception {
- return (retainedTsBlock == null)
- && (currentDeviceIndex >= deviceCount || seriesScanOptions.limitConsumedUp());
- }
-
- @Override
- public long calculateMaxPeekMemory() {
- return (1L + columnsIndexArray.length)
- * TSFileDescriptor.getInstance().getConfig().getPageSizeInByte();
- }
-
- @Override
- public List<TSDataType> getResultDataTypes() {
- List<TSDataType> resultDataTypes = new ArrayList<>(columnSchemas.size());
- for (ColumnSchema columnSchema : columnSchemas) {
- resultDataTypes.add(getTSDataType(columnSchema.getType()));
- }
- return resultDataTypes;
- }
-
- @Override
- public void initQueryDataSource(IQueryDataSource dataSource) {
- this.queryDataSource = (QueryDataSource) dataSource;
- if (this.seriesScanUtil != null) {
- this.seriesScanUtil.initQueryDataSource(queryDataSource);
- }
- this.resultTsBlockBuilder = new TsBlockBuilder(getResultDataTypes());
- this.resultTsBlockBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum);
- this.measurementDataBuilder = new TsBlockBuilder(this.measurementColumnTSDataTypes);
- this.measurementDataBuilder.setMaxTsBlockLineNumber(this.maxTsBlockLineNum);
- }
-
- private void prepareForNextDevice() {
- if (currentDeviceIndex < deviceCount) {
- // construct AlignedSeriesScanUtil for next device
- constructAlignedSeriesScanUtil();
-
- // reset QueryDataSource
- queryDataSource.reset();
- this.seriesScanUtil.initQueryDataSource(queryDataSource);
- this.operatorContext.recordSpecifiedInfo(
- CURRENT_DEVICE_INDEX_STRING, Integer.toString(currentDeviceIndex));
- }
- }
-
- private void constructAlignedSeriesScanUtil() {
- if (this.deviceEntries.isEmpty()) {
- // no need to construct SeriesScanUtil, hasNext will return false
- return;
- }
-
- if (this.deviceEntries.get(this.currentDeviceIndex) == null) {
- throw new IllegalStateException(
- "Device entries of index " + this.currentDeviceIndex + " in TableScanOperator is empty");
- }
-
- DeviceEntry deviceEntry = this.deviceEntries.get(this.currentDeviceIndex);
- AlignedFullPath alignedPath =
- constructAlignedPath(deviceEntry, measurementColumnNames, measurementSchemas, allSensors);
- this.seriesScanUtil =
- new AlignedSeriesScanUtil(
- alignedPath,
- scanOrder,
- seriesScanOptions,
- operatorContext.getInstanceContext(),
- true,
- measurementColumnTSDataTypes);
- }
-
- public static AlignedFullPath constructAlignedPath(
- DeviceEntry deviceEntry,
- List<String> measurementColumnNames,
- List<IMeasurementSchema> measurementSchemas,
- Set<String> allSensors) {
- return new AlignedFullPath(
- deviceEntry.getDeviceID(), measurementColumnNames, measurementSchemas, allSensors);
+public class TableScanOperator extends AbstractTableScanOperator {
+ public TableScanOperator(AbstractTableScanOperatorParameter parameter) {
+ super(parameter);
}
@Override
- public long ramBytesUsed() {
- return INSTANCE_SIZE
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(seriesScanUtil)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(operatorContext)
- + MemoryEstimationHelper.getEstimatedSizeOfAccountableObject(sourceId)
- + (resultTsBlockBuilder == null ? 0 : resultTsBlockBuilder.getRetainedSizeInBytes())
- + RamUsageEstimator.sizeOfCollection(deviceEntries);
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ // +1 for skipping the table name segment
+ return ((String) deviceEntry.getNthSegment(idColumnIndex + 1));
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewAggregationScanOperator.java
similarity index 51%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewAggregationScanOperator.java
index 9ab432c1e35..4bc073e63dc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewAggregationScanOperator.java
@@ -17,20 +17,26 @@
* under the License.
*/
-package org.apache.iotdb.db.queryengine.plan.planner.plan.node.source;
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.IPartitionRelatedNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
-public abstract class SourceNode extends PlanNode implements AutoCloseable, IPartitionRelatedNode {
+import org.apache.tsfile.file.metadata.IDeviceID;
- protected SourceNode(final PlanNodeId id) {
- super(id);
- }
+public class TreeAlignedDeviceViewAggregationScanOperator
+ extends AbstractDefaultAggTableScanOperator {
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
- public abstract void open() throws Exception;
+ public TreeAlignedDeviceViewAggregationScanOperator(
+ AbstractAggTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor) {
+ super(parameter);
+ this.extractor = extractor;
+ }
- public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewScanOperator.java
similarity index 52%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewScanOperator.java
index 9ab432c1e35..03d0e998e71 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/TreeAlignedDeviceViewScanOperator.java
@@ -17,20 +17,25 @@
* under the License.
*/
-package org.apache.iotdb.db.queryengine.plan.planner.plan.node.source;
+package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.IPartitionRelatedNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
-public abstract class SourceNode extends PlanNode implements AutoCloseable, IPartitionRelatedNode {
+import org.apache.tsfile.file.metadata.IDeviceID;
- protected SourceNode(final PlanNodeId id) {
- super(id);
- }
+public class TreeAlignedDeviceViewScanOperator extends AbstractTableScanOperator {
+
+ private final IDeviceID.TreeDeviceIdColumnValueExtractor extractor;
- public abstract void open() throws Exception;
+ public TreeAlignedDeviceViewScanOperator(
+ AbstractTableScanOperatorParameter parameter,
+ IDeviceID.TreeDeviceIdColumnValueExtractor extractor) {
+ super(parameter);
+ this.extractor = extractor;
+ }
- public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
+ @Override
+ String getNthIdColumnValue(DeviceEntry deviceEntry, int idColumnIndex) {
+ return (String) extractor.extract(deviceEntry.getDeviceID(), idColumnIndex);
+ }
}
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 741e05a64bd..d1cf042ef52 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
@@ -21,7 +21,9 @@ package org.apache.iotdb.db.queryengine.plan.planner;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.path.AlignedFullPath;
+import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.schema.column.ColumnHeader;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
@@ -72,14 +74,18 @@ import org.apache.iotdb.db.queryengine.execution.operator.schema.SchemaQueryScan
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.DevicePredicateFilter;
import org.apache.iotdb.db.queryengine.execution.operator.schema.source.SchemaSourceFactory;
import org.apache.iotdb.db.queryengine.execution.operator.sink.IdentitySinkOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.AbstractDataSourceOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.ExchangeOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractAggTableScanOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.AbstractTableScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.DefaultAggTableScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.InformationSchemaTableScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.LastQueryAggTableScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.MergeSortFullOuterJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.MergeSortInnerJoinOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.TableScanOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.TreeAlignedDeviceViewAggregationScanOperator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.TreeAlignedDeviceViewScanOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AggregationOperator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.LastByDescAccumulator;
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.LastDescAccumulator;
@@ -120,6 +126,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.SortOrder;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
@@ -139,6 +146,8 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceFetchNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceQueryCountNode;
@@ -162,6 +171,10 @@ import org.apache.tsfile.block.column.Column;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.common.conf.TSFileDescriptor;
import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.idcolumn.FourOrHigherLevelDBExtractor;
+import org.apache.tsfile.file.metadata.idcolumn.ThreeLevelDBExtractor;
+import org.apache.tsfile.file.metadata.idcolumn.TwoLevelDBExtractor;
import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.common.TimeRange;
import org.apache.tsfile.read.common.block.column.BinaryColumn;
@@ -330,8 +343,95 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
}
@Override
- public Operator visitDeviceTableScan(
- DeviceTableScanNode node, LocalExecutionPlanContext context) {
+ public Operator visitTreeNonAlignedDeviceViewScan(
+ TreeNonAlignedDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ throw new UnsupportedOperationException(
+ "view for non aligned devices in tree is not supported");
+ }
+
+ public static IDeviceID.TreeDeviceIdColumnValueExtractor createTreeDeviceIdColumnValueExtractor(
+ String treeDBName) {
+ try {
+ PartialPath db = new PartialPath(treeDBName);
+ int dbLevel = db.getNodes().length;
+ if (dbLevel == 2) {
+ return new TwoLevelDBExtractor(treeDBName.length());
+ } else if (dbLevel == 3) {
+ return new ThreeLevelDBExtractor(treeDBName.length());
+ } else if (dbLevel >= 4) {
+ return new FourOrHigherLevelDBExtractor(dbLevel);
+ } else {
+ throw new IllegalArgumentException(
+ "tree db name should at least be two level: " + treeDBName);
+ }
+ } catch (IllegalPathException e) {
+ throw new IllegalArgumentException(e);
+ }
+ }
+
+ @Override
+ public Operator visitTreeAlignedDeviceViewScan(
+ TreeAlignedDeviceViewScanNode node, LocalExecutionPlanContext context) {
+
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(node.getTreeDBName());
+
+ AbstractTableScanOperator.AbstractTableScanOperatorParameter parameter =
+ constructAbstractTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap());
+
+ TreeAlignedDeviceViewScanOperator treeAlignedDeviceViewScanOperator =
+ new TreeAlignedDeviceViewScanOperator(parameter, idColumnValueExtractor);
+
+ addSource(
+ treeAlignedDeviceViewScanOperator,
+ context,
+ node,
+ parameter.measurementColumnNames,
+ parameter.measurementSchemas,
+ parameter.allSensors,
+ TreeAlignedDeviceViewScanNode.class.getSimpleName());
+
+ return treeAlignedDeviceViewScanOperator;
+ }
+
+ private void addSource(
+ AbstractDataSourceOperator sourceOperator,
+ LocalExecutionPlanContext context,
+ DeviceTableScanNode node,
+ List<String> measurementColumnNames,
+ List<IMeasurementSchema> measurementSchemas,
+ Set<String> allSensors,
+ String planNodeName) {
+
+ ((DataDriverContext) context.getDriverContext()).addSourceOperator(sourceOperator);
+
+ for (int i = 0, size = node.getDeviceEntries().size(); i < size; i++) {
+ if (node.getDeviceEntries().get(i) == null) {
+ throw new IllegalStateException(
+ "Device entries of index " + i + " in " + planNodeName + " is empty");
+ }
+ AlignedFullPath alignedPath =
+ constructAlignedPath(
+ node.getDeviceEntries().get(i),
+ measurementColumnNames,
+ measurementSchemas,
+ allSensors);
+ ((DataDriverContext) context.getDriverContext()).addPath(alignedPath);
+ }
+
+ context.getDriverContext().setInputDriver(true);
+ }
+
+ private AbstractTableScanOperator.AbstractTableScanOperatorParameter
+ constructAbstractTableScanOperatorParameter(
+ DeviceTableScanNode node,
+ LocalExecutionPlanContext context,
+ String className,
+ Map<String, String> fieldColumnsRenameMap) {
List<Symbol> outputColumnNames = node.getOutputSymbols();
int outputColumnCount = outputColumnNames.size();
@@ -360,9 +460,13 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
case FIELD:
columnsIndexArray[idx++] = measurementColumnCount;
measurementColumnCount++;
- measurementColumnNames.add(schema.getName());
+
+ String realMeasurementName =
+ fieldColumnsRenameMap.getOrDefault(schema.getName(), schema.getName());
+
+ measurementColumnNames.add(realMeasurementName);
measurementSchemas.add(
- new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ new MeasurementSchema(realMeasurementName, getTSDataType(schema.getType())));
columnSchemas.add(schema);
measurementColumnsIndexMap.put(columnName.getName(), measurementColumnCount - 1);
break;
@@ -381,10 +485,13 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
for (Map.Entry<Symbol, ColumnSchema> entry : node.getAssignments().entrySet()) {
if (!outputSet.contains(entry.getKey()) && entry.getValue().getColumnCategory() == FIELD) {
measurementColumnCount++;
- measurementColumnNames.add(entry.getValue().getName());
+ String realMeasurementName =
+ fieldColumnsRenameMap.getOrDefault(
+ entry.getValue().getName(), entry.getValue().getName());
+
+ measurementColumnNames.add(realMeasurementName);
measurementSchemas.add(
- new MeasurementSchema(
- entry.getValue().getName(), getTSDataType(entry.getValue().getType())));
+ new MeasurementSchema(realMeasurementName, getTSDataType(entry.getValue().getType())));
measurementColumnsIndexMap.put(entry.getKey().getName(), measurementColumnCount - 1);
} else if (entry.getValue().getColumnCategory() == TIME) {
timeColumnName = entry.getKey().getName();
@@ -407,10 +514,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
OperatorContext operatorContext =
context
.getDriverContext()
- .addOperatorContext(
- context.getNextOperatorId(),
- node.getPlanNodeId(),
- TableScanOperator.class.getSimpleName());
+ .addOperatorContext(context.getNextOperatorId(), node.getPlanNodeId(), className);
int maxTsBlockLineNum = TSFileDescriptor.getInstance().getConfig().getMaxTsBlockLineNumber();
if (context.getTypeProvider().getTemplatedInfo() != null) {
@@ -423,33 +527,46 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
Set<String> allSensors = new HashSet<>(measurementColumnNames);
// for time column
allSensors.add("");
- TableScanOperator tableScanOperator =
- new TableScanOperator(
- operatorContext,
- node.getPlanNodeId(),
- columnSchemas,
- columnsIndexArray,
- node.getDeviceEntries(),
- node.getScanOrder(),
- seriesScanOptions,
- measurementColumnNames,
- allSensors,
- measurementSchemas,
- maxTsBlockLineNum);
- ((DataDriverContext) context.getDriverContext()).addSourceOperator(tableScanOperator);
+ return new AbstractTableScanOperator.AbstractTableScanOperatorParameter(
+ allSensors,
+ operatorContext,
+ node.getPlanNodeId(),
+ columnSchemas,
+ columnsIndexArray,
+ node.getDeviceEntries(),
+ node.getScanOrder(),
+ seriesScanOptions,
+ measurementColumnNames,
+ measurementSchemas,
+ maxTsBlockLineNum);
+ }
- for (int i = 0, size = node.getDeviceEntries().size(); i < size; i++) {
- AlignedFullPath alignedPath =
- constructAlignedPath(
- node.getDeviceEntries().get(i),
- measurementColumnNames,
- measurementSchemas,
- allSensors);
- ((DataDriverContext) context.getDriverContext()).addPath(alignedPath);
- }
+ // used for TableScanOperator
+ private AbstractTableScanOperator.AbstractTableScanOperatorParameter
+ constructAbstractTableScanOperatorParameter(
+ DeviceTableScanNode node, LocalExecutionPlanContext context) {
+ return constructAbstractTableScanOperatorParameter(
+ node, context, TableScanOperator.class.getSimpleName(), Collections.emptyMap());
+ }
- context.getDriverContext().setInputDriver(true);
+ @Override
+ public Operator visitDeviceTableScan(
+ DeviceTableScanNode node, LocalExecutionPlanContext context) {
+
+ AbstractTableScanOperator.AbstractTableScanOperatorParameter parameter =
+ constructAbstractTableScanOperatorParameter(node, context);
+
+ TableScanOperator tableScanOperator = new TableScanOperator(parameter);
+
+ addSource(
+ tableScanOperator,
+ context,
+ node,
+ parameter.measurementColumnNames,
+ parameter.measurementSchemas,
+ parameter.allSensors,
+ DeviceTableScanNode.class.getSimpleName());
return tableScanOperator;
}
@@ -1755,8 +1872,38 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
}
@Override
- public Operator visitAggregationTableScan(
- AggregationTableScanNode node, LocalExecutionPlanContext context) {
+ public Operator visitAggregationTreeDeviceViewScan(
+ AggregationTreeDeviceViewScanNode node, LocalExecutionPlanContext context) {
+ IDeviceID.TreeDeviceIdColumnValueExtractor idColumnValueExtractor =
+ createTreeDeviceIdColumnValueExtractor(node.getTreeDBName());
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(
+ node,
+ context,
+ TreeAlignedDeviceViewAggregationScanOperator.class.getSimpleName(),
+ node.getMeasurementColumnNameMap());
+
+ TreeAlignedDeviceViewAggregationScanOperator treeAlignedDeviceViewAggregationScanOperator =
+ new TreeAlignedDeviceViewAggregationScanOperator(parameter, idColumnValueExtractor);
+
+ addSource(
+ treeAlignedDeviceViewAggregationScanOperator,
+ context,
+ node,
+ parameter.getMeasurementColumnNames(),
+ parameter.getMeasurementSchemas(),
+ parameter.getAllSensors(),
+ AggregationTreeDeviceViewScanNode.class.getSimpleName());
+ return treeAlignedDeviceViewAggregationScanOperator;
+ }
+
+ private AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter
+ constructAbstractAggTableScanOperatorParameter(
+ AggregationTableScanNode node,
+ LocalExecutionPlanContext context,
+ String className,
+ Map<String, String> fieldColumnsRenameMap) {
List<String> measurementColumnNames = new ArrayList<>();
List<IMeasurementSchema> measurementSchemas = new ArrayList<>();
@@ -1798,9 +1945,11 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
case FIELD:
aggColumnsIndexArray[channel] = measurementColumnCount;
measurementColumnCount++;
- measurementColumnNames.add(schema.getName());
+ String realMeasurementName =
+ fieldColumnsRenameMap.getOrDefault(schema.getName(), schema.getName());
+ measurementColumnNames.add(realMeasurementName);
measurementSchemas.add(
- new MeasurementSchema(schema.getName(), getTSDataType(schema.getType())));
+ new MeasurementSchema(realMeasurementName, getTSDataType(schema.getType())));
measurementColumnsIndexMap.put(symbol.getName(), measurementColumnCount - 1);
break;
case TIME:
@@ -1825,10 +1974,12 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
if (!aggColumnLayout.containsKey(entry.getKey())
&& entry.getValue().getColumnCategory() == FIELD) {
measurementColumnCount++;
- measurementColumnNames.add(entry.getValue().getName());
+ String realMeasurementName =
+ fieldColumnsRenameMap.getOrDefault(
+ entry.getValue().getName(), entry.getValue().getName());
+ measurementColumnNames.add(realMeasurementName);
measurementSchemas.add(
- new MeasurementSchema(
- entry.getValue().getName(), getTSDataType(entry.getValue().getType())));
+ new MeasurementSchema(realMeasurementName, getTSDataType(entry.getValue().getType())));
measurementColumnsIndexMap.put(entry.getKey().getName(), measurementColumnCount - 1);
} else if (entry.getValue().getColumnCategory() == TIME) {
timeColumnName = entry.getKey().getName();
@@ -1899,10 +2050,7 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
final OperatorContext operatorContext =
context
.getDriverContext()
- .addOperatorContext(
- context.getNextOperatorId(),
- node.getPlanNodeId(),
- AbstractAggTableScanOperator.class.getSimpleName());
+ .addOperatorContext(context.getNextOperatorId(), node.getPlanNodeId(), className);
SeriesScanOptions seriesScanOptions =
buildSeriesScanOptions(
context,
@@ -1920,122 +2068,134 @@ public class TableOperatorGenerator extends PlanVisitor<Operator, LocalExecution
allSensors.add(""); // for time column
context.getDriverContext().setInputDriver(true);
- if (canUseLastCacheOptimize(aggregators, node, timeColumnName)) {
- List<Integer> hitCachesIndexes = new ArrayList<>();
- List<Pair<OptionalLong, TsPrimitiveType[]>> hitCachedResults = new ArrayList<>();
- List<DeviceEntry> cachedDeviceEntries = new ArrayList<>();
- List<DeviceEntry> unCachedDeviceEntries = new ArrayList<>();
- long tableTTL =
- DataNodeTTLCache.getInstance()
- .getTTLForTable(
+ return new AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter(
+ node.getPlanNodeId(),
+ operatorContext,
+ aggColumnSchemas,
+ aggColumnsIndexArray,
+ node.getDeviceEntries(),
+ node.getDeviceEntries().size(),
+ seriesScanOptions,
+ measurementColumnNames,
+ allSensors,
+ measurementSchemas,
+ aggregators,
+ groupingKeySchemas,
+ groupingKeyIndex,
+ timeRangeIterator,
+ scanAscending,
+ canUseStatistic,
+ aggregatorInputChannels,
+ timeColumnName);
+ }
+
+ // used for AggregationTableScanNode
+ private AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter
+ constructAbstractAggTableScanOperatorParameter(
+ AggregationTableScanNode node, LocalExecutionPlanContext context) {
+ return constructAbstractAggTableScanOperatorParameter(
+ node, context, AbstractAggTableScanOperator.class.getSimpleName(), Collections.emptyMap());
+ }
+
+ @Override
+ public Operator visitAggregationTableScan(
+ AggregationTableScanNode node, LocalExecutionPlanContext context) {
+
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter =
+ constructAbstractAggTableScanOperatorParameter(node, context);
+
+ if (canUseLastCacheOptimize(
+ parameter.getTableAggregators(), node, parameter.getTimeColumnName())) {
+ return constructLastQueryAggTableScanOperator(node, parameter, context);
+ } else {
+ DefaultAggTableScanOperator aggTableScanOperator = new DefaultAggTableScanOperator(parameter);
+
+ addSource(
+ aggTableScanOperator,
+ context,
+ node,
+ parameter.getMeasurementColumnNames(),
+ parameter.getMeasurementSchemas(),
+ parameter.getAllSensors(),
+ AggregationTableScanNode.class.getSimpleName());
+ return aggTableScanOperator;
+ }
+ }
+
+ private LastQueryAggTableScanOperator constructLastQueryAggTableScanOperator(
+ AggregationTableScanNode node,
+ AbstractAggTableScanOperator.AbstractAggTableScanOperatorParameter parameter,
+ LocalExecutionPlanContext context) {
+ List<Integer> hitCachesIndexes = new ArrayList<>();
+ List<Pair<OptionalLong, TsPrimitiveType[]>> hitCachedResults = new ArrayList<>();
+ List<DeviceEntry> cachedDeviceEntries = new ArrayList<>();
+ List<DeviceEntry> unCachedDeviceEntries = new ArrayList<>();
+ long tableTTL =
+ DataNodeTTLCache.getInstance()
+ .getTTLForTable(
+ node.getQualifiedObjectName().getDatabaseName(),
+ node.getQualifiedObjectName().getObjectName());
+ Filter updateTimeFilter =
+ updateFilterUsingTTL(parameter.getSeriesScanOptions().getGlobalTimeFilter(), tableTTL);
+ for (int i = 0; i < node.getDeviceEntries().size(); i++) {
+ Optional<Pair<OptionalLong, TsPrimitiveType[]>> lastByResult =
+ TableDeviceSchemaCache.getInstance()
+ .getLastRow(
node.getQualifiedObjectName().getDatabaseName(),
- node.getQualifiedObjectName().getObjectName());
- Filter updateTimeFilter =
- updateFilterUsingTTL(seriesScanOptions.getGlobalTimeFilter(), tableTTL);
- for (int i = 0; i < node.getDeviceEntries().size(); i++) {
- Optional<Pair<OptionalLong, TsPrimitiveType[]>> lastByResult =
- TableDeviceSchemaCache.getInstance()
- .getLastRow(
- node.getQualifiedObjectName().getDatabaseName(),
- node.getDeviceEntries().get(i).getDeviceID(),
- "",
- measurementColumnNames);
- boolean allHitCache = true;
- if (lastByResult.isPresent() && lastByResult.get().getLeft().isPresent()) {
- for (int j = 0; j < lastByResult.get().getRight().length; j++) {
- TsPrimitiveType tsPrimitiveType = lastByResult.get().getRight()[j];
- if (tsPrimitiveType == null
- || (updateTimeFilter != null
- && !LastQueryUtil.satisfyFilter(
- updateTimeFilter,
- new TimeValuePair(
- lastByResult.get().getLeft().getAsLong(), tsPrimitiveType)))) {
- // the process logic is different from tree model which examine if
- // `isFilterGtOrGe(seriesScanOptions.getGlobalTimeFilter())`, set
- // `lastByResult.get().getRight()[j] = EMPTY_PRIMITIVE_TYPE`,
- // but it should skip in table model
- allHitCache = false;
- break;
- }
+ node.getDeviceEntries().get(i).getDeviceID(),
+ "",
+ parameter.getMeasurementColumnNames());
+ boolean allHitCache = true;
+ if (lastByResult.isPresent() && lastByResult.get().getLeft().isPresent()) {
+ for (int j = 0; j < lastByResult.get().getRight().length; j++) {
+ TsPrimitiveType tsPrimitiveType = lastByResult.get().getRight()[j];
+ if (tsPrimitiveType == null
+ || (updateTimeFilter != null
+ && !LastQueryUtil.satisfyFilter(
+ updateTimeFilter,
+ new TimeValuePair(
+ lastByResult.get().getLeft().getAsLong(), tsPrimitiveType)))) {
+ // the process logic is different from tree model which examine if
+ // `isFilterGtOrGe(seriesScanOptions.getGlobalTimeFilter())`, set
+ // `lastByResult.get().getRight()[j] = EMPTY_PRIMITIVE_TYPE`,
+ // but it should skip in table model
+ allHitCache = false;
+ break;
}
- } else {
- allHitCache = false;
- }
-
- if (!allHitCache) {
- AlignedFullPath alignedPath =
- constructAlignedPath(
- node.getDeviceEntries().get(i),
- measurementColumnNames,
- measurementSchemas,
- allSensors);
- ((DataDriverContext) context.getDriverContext()).addPath(alignedPath);
- unCachedDeviceEntries.add(node.getDeviceEntries().get(i));
- } else {
- hitCachesIndexes.add(i);
- hitCachedResults.add(lastByResult.get());
- cachedDeviceEntries.add(node.getDeviceEntries().get(i));
}
+ } else {
+ allHitCache = false;
}
- // context add TableLastQueryOperator
- LastQueryAggTableScanOperator lastQueryOperator =
- new LastQueryAggTableScanOperator(
- node.getPlanNodeId(),
- operatorContext,
- aggColumnSchemas,
- aggColumnsIndexArray,
- unCachedDeviceEntries,
- cachedDeviceEntries,
- seriesScanOptions,
- measurementColumnNames,
- allSensors,
- measurementSchemas,
- aggregators,
- groupingKeySchemas,
- groupingKeyIndex,
- timeRangeIterator,
- scanAscending,
- canUseStatistic,
- aggregatorInputChannels,
- node.getQualifiedObjectName(),
- hitCachesIndexes,
- hitCachedResults);
-
- ((DataDriverContext) context.getDriverContext()).addSourceOperator(lastQueryOperator);
- return lastQueryOperator;
- } else {
- DefaultAggTableScanOperator aggTableScanOperator =
- new DefaultAggTableScanOperator(
- node.getPlanNodeId(),
- operatorContext,
- aggColumnSchemas,
- aggColumnsIndexArray,
- node.getDeviceEntries(),
- node.getDeviceEntries().size(),
- seriesScanOptions,
- measurementColumnNames,
- allSensors,
- measurementSchemas,
- aggregators,
- groupingKeySchemas,
- groupingKeyIndex,
- timeRangeIterator,
- scanAscending,
- canUseStatistic,
- aggregatorInputChannels);
- for (int i = 0; i < node.getDeviceEntries().size(); i++) {
+ if (!allHitCache) {
AlignedFullPath alignedPath =
constructAlignedPath(
node.getDeviceEntries().get(i),
- measurementColumnNames,
- measurementSchemas,
- allSensors);
+ parameter.getMeasurementColumnNames(),
+ parameter.getMeasurementSchemas(),
+ parameter.getAllSensors());
((DataDriverContext) context.getDriverContext()).addPath(alignedPath);
+ unCachedDeviceEntries.add(node.getDeviceEntries().get(i));
+ } else {
+ hitCachesIndexes.add(i);
+ hitCachedResults.add(lastByResult.get());
+ cachedDeviceEntries.add(node.getDeviceEntries().get(i));
}
- ((DataDriverContext) context.getDriverContext()).addSourceOperator(aggTableScanOperator);
- return aggTableScanOperator;
}
+
+ parameter.setDeviceEntries(unCachedDeviceEntries);
+
+ // context add TableLastQueryOperator
+ LastQueryAggTableScanOperator lastQueryOperator =
+ new LastQueryAggTableScanOperator(
+ parameter,
+ cachedDeviceEntries,
+ node.getQualifiedObjectName(),
+ hitCachesIndexes,
+ hitCachedResults);
+
+ ((DataDriverContext) context.getDriverContext()).addSourceOperator(lastQueryOperator);
+ return lastQueryOperator;
}
private SeriesScanOptions buildSeriesScanOptions(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
index 48a339dcb0d..1811d7f46ae 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanGraphPrinter.java
@@ -65,6 +65,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationDe
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.CrossSeriesAggregationDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.DeviceViewIntoPathDescriptor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.IntoPathDescriptor;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ExchangeNode;
@@ -72,6 +73,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.commons.lang3.Validate;
@@ -628,7 +630,7 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
}
List<String> boxValue = new ArrayList<>();
- boxValue.add(String.format("TableScan-%s", node.getPlanNodeId().getId()));
+ boxValue.add(node.toString());
boxValue.add(String.format("QualifiedTableName: %s", node.getQualifiedObjectName().toString()));
boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
@@ -660,6 +662,14 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
node.getRegionReplicaSet() == null || node.getRegionReplicaSet().getRegionId() == null
? REGION_NOT_ASSIGNED
: node.getRegionReplicaSet().getRegionId().getId()));
+
+ if (deviceTableScanNode instanceof TreeDeviceViewScanNode) {
+ TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) deviceTableScanNode;
+ boxValue.add(String.format("TreeDB: %s", treeDeviceViewScanNode.getTreeDBName()));
+ boxValue.add(
+ String.format(
+ "MeasurementToColumnName: %s", treeDeviceViewScanNode.getMeasurementColumnNameMap()));
+ }
return render(node, boxValue, context);
}
@@ -690,7 +700,7 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode node,
GraphContext context) {
List<String> boxValue = new ArrayList<>();
- boxValue.add(String.format("AggregationTableScan-%s", node.getPlanNodeId().getId()));
+ boxValue.add(node.toString());
boxValue.add(String.format("QualifiedTableName: %s", node.getQualifiedObjectName().toString()));
boxValue.add(String.format("OutputSymbols: %s", node.getOutputSymbols()));
int i = 0;
@@ -725,6 +735,16 @@ public class PlanGraphPrinter extends PlanVisitor<List<String>, PlanGraphPrinter
node.getRegionReplicaSet() == null || node.getRegionReplicaSet().getRegionId() == null
? REGION_NOT_ASSIGNED
: node.getRegionReplicaSet().getRegionId().getId()));
+
+ if (node instanceof AggregationTreeDeviceViewScanNode) {
+ AggregationTreeDeviceViewScanNode aggregationTreeDeviceViewScanNode =
+ (AggregationTreeDeviceViewScanNode) node;
+ boxValue.add(String.format("TreeDB: %s", aggregationTreeDeviceViewScanNode.getTreeDBName()));
+ boxValue.add(
+ String.format(
+ "MeasurementToColumnName: %s",
+ aggregationTreeDeviceViewScanNode.getMeasurementColumnNameMap()));
+ }
return render(node, boxValue, context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
index 834bd339c0e..4e3196c8d42 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNode.java
@@ -51,6 +51,8 @@ public abstract class PlanNode implements IConsensusRequest {
protected boolean isGeneratedByPipe = false;
+ protected PlanNode() {}
+
protected PlanNode(PlanNodeId id) {
requireNonNull(id, "id is null");
this.id = id;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
index dd67e129360..9a8e5ed31e6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanNodeType.java
@@ -116,11 +116,14 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDe
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
@@ -282,6 +285,9 @@ public enum PlanNodeType {
TABLE_EXPLAIN_ANALYZE_NODE((short) 1019),
TABLE_ENFORCE_SINGLE_ROW_NODE((short) 1020),
INFORMATION_SCHEMA_TABLE_SCAN_NODE((short) 1021),
+ AGGREGATION_TREE_DEVICE_VIEW_SCAN_NODE((short) 1022),
+ TREE_ALIGNED_DEVICE_VIEW_SCAN_NODE((short) 1023),
+ TREE_NONALIGNED_DEVICE_VIEW_SCAN_NODE((short) 1024),
RELATIONAL_INSERT_TABLET((short) 2000),
RELATIONAL_INSERT_ROW((short) 2001),
@@ -641,6 +647,12 @@ public enum PlanNodeType {
return EnforceSingleRowNode.deserialize(buffer);
case 1021:
return InformationSchemaTableScanNode.deserialize(buffer);
+ case 1022:
+ return AggregationTreeDeviceViewScanNode.deserialize(buffer);
+ case 1023:
+ return TreeAlignedDeviceViewScanNode.deserialize(buffer);
+ case 1024:
+ return TreeNonAlignedDeviceViewScanNode.deserialize(buffer);
case 2000:
return RelationalInsertTabletNode.deserialize(buffer);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
index 9c172b53b1c..348104f95f9 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/PlanVisitor.java
@@ -121,12 +121,16 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalIn
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.GroupReference;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.GapFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.LinearFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.ConstructTableDevicesBlackListNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.CreateOrUpdateTableDeviceNode;
@@ -770,4 +774,20 @@ public abstract class PlanVisitor<R, C> {
C context) {
return visitDeviceTableScan(node, context);
}
+
+ public R visitTreeDeviceViewScan(TreeDeviceViewScanNode node, C context) {
+ return visitDeviceTableScan(node, context);
+ }
+
+ public R visitAggregationTreeDeviceViewScan(AggregationTreeDeviceViewScanNode node, C context) {
+ return visitAggregationTableScan(node, context);
+ }
+
+ public R visitTreeAlignedDeviceViewScan(TreeAlignedDeviceViewScanNode node, C context) {
+ return visitTreeDeviceViewScan(node, context);
+ }
+
+ public R visitTreeNonAlignedDeviceViewScan(TreeNonAlignedDeviceViewScanNode node, C context) {
+ return visitTreeDeviceViewScan(node, context);
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
index 9ab432c1e35..bf930d1d63b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
public abstract class SourceNode extends PlanNode implements AutoCloseable, IPartitionRelatedNode {
+ protected SourceNode() {}
protected SourceNode(final PlanNodeId id) {
super(id);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/AlignedDeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/AlignedDeviceEntry.java
new file mode 100644
index 00000000000..fcb9375dd55
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/AlignedDeviceEntry.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.plan.relational.metadata;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.Binary;
+
+import java.util.List;
+import java.util.Objects;
+
+public class AlignedDeviceEntry extends DeviceEntry {
+
+ public AlignedDeviceEntry(IDeviceID deviceID, List<Binary> attributeColumnValues) {
+ super(deviceID, attributeColumnValues);
+ }
+
+ @Override
+ public String toString() {
+ return "AlignedDeviceEntry{"
+ + "deviceID="
+ + deviceID
+ + ", attributeColumnValues="
+ + attributeColumnValues
+ + '}';
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final AlignedDeviceEntry that = (AlignedDeviceEntry) obj;
+ return Objects.equals(deviceID, that.deviceID)
+ && Objects.equals(attributeColumnValues, that.attributeColumnValues);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
index eb06b88e127..f6d8f594d9a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/DeviceEntry.java
@@ -45,13 +45,13 @@ import static org.apache.tsfile.utils.ReadWriteIOUtils.write;
* {@code null}s are trimmed thus will not appear in the {@link IDeviceID}, and it will be like
* "a.b".
*/
-public class DeviceEntry implements Accountable {
+public abstract class DeviceEntry implements Accountable {
private static final long INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(DeviceEntry.class);
- private final IDeviceID deviceID;
- private final List<Binary> attributeColumnValues;
+ protected final IDeviceID deviceID;
+ protected final List<Binary> attributeColumnValues;
public DeviceEntry(final IDeviceID deviceID, final List<Binary> attributeColumnValues) {
this.deviceID = deviceID;
@@ -78,6 +78,11 @@ public class DeviceEntry implements Accountable {
for (final Binary value : attributeColumnValues) {
serializeBinary(byteBuffer, value);
}
+ write(
+ this instanceof AlignedDeviceEntry
+ ? DeviceEntryType.ALIGNED.ordinal()
+ : DeviceEntryType.NON_ALIGNED.ordinal(),
+ byteBuffer);
}
public void serialize(final DataOutputStream stream) throws IOException {
@@ -86,6 +91,11 @@ public class DeviceEntry implements Accountable {
for (final Binary value : attributeColumnValues) {
serializeBinary(stream, value);
}
+ write(
+ this instanceof AlignedDeviceEntry
+ ? DeviceEntryType.ALIGNED.ordinal()
+ : DeviceEntryType.NON_ALIGNED.ordinal(),
+ stream);
}
public static DeviceEntry deserialize(final ByteBuffer byteBuffer) {
@@ -95,7 +105,7 @@ public class DeviceEntry implements Accountable {
while (size-- > 0) {
attributeColumnValues.add(deserializeBinary(byteBuffer));
}
- return new DeviceEntry(iDeviceID, attributeColumnValues);
+ return constructDeviceEntry(iDeviceID, attributeColumnValues, readInt(byteBuffer));
}
public static void serializeBinary(final ByteBuffer byteBuffer, final Binary binary) {
@@ -124,14 +134,22 @@ public class DeviceEntry implements Accountable {
return new Binary(bytes);
}
- @Override
- public String toString() {
- return "DeviceEntry{"
- + "deviceID="
- + deviceID
- + ", attributeColumnValues="
- + attributeColumnValues
- + '}';
+ public enum DeviceEntryType {
+ ALIGNED,
+ NON_ALIGNED
+ }
+
+ private static DeviceEntry constructDeviceEntry(
+ IDeviceID deviceID, List<Binary> attributeColumnValues, int ordinal) {
+ switch (DeviceEntryType.values()[ordinal]) {
+ case ALIGNED:
+ return new AlignedDeviceEntry(deviceID, attributeColumnValues);
+ case NON_ALIGNED:
+ return new NonAlignedAlignedDeviceEntry(deviceID, attributeColumnValues);
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown AlignedDeviceEntry Type: " + DeviceEntryType.values()[ordinal]);
+ }
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/NonAlignedAlignedDeviceEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/NonAlignedAlignedDeviceEntry.java
new file mode 100644
index 00000000000..9c6a2863922
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/NonAlignedAlignedDeviceEntry.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.plan.relational.metadata;
+
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.utils.Binary;
+
+import java.util.List;
+import java.util.Objects;
+
+public class NonAlignedAlignedDeviceEntry extends DeviceEntry {
+
+ public NonAlignedAlignedDeviceEntry(IDeviceID deviceID, List<Binary> attributeColumnValues) {
+ super(deviceID, attributeColumnValues);
+ }
+
+ @Override
+ public String toString() {
+ return "NonAlignedAlignedDeviceEntry{"
+ + "deviceID="
+ + deviceID
+ + ", attributeColumnValues="
+ + attributeColumnValues
+ + '}';
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final NonAlignedAlignedDeviceEntry that = (NonAlignedAlignedDeviceEntry) obj;
+ return Objects.equals(deviceID, that.deviceID)
+ && Objects.equals(attributeColumnValues, that.attributeColumnValues);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TreeDeviceViewSchema.java
similarity index 55%
copy from iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
copy to iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TreeDeviceViewSchema.java
index 9ab432c1e35..08f88d7629c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/source/SourceNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/TreeDeviceViewSchema.java
@@ -17,20 +17,23 @@
* under the License.
*/
-package org.apache.iotdb.db.queryengine.plan.planner.plan.node.source;
+package org.apache.iotdb.db.queryengine.plan.relational.metadata;
-import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.IPartitionRelatedNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
-import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
-public abstract class SourceNode extends PlanNode implements AutoCloseable, IPartitionRelatedNode {
+public class TreeDeviceViewSchema extends TableSchema {
- protected SourceNode(final PlanNodeId id) {
- super(id);
+ public TreeDeviceViewSchema(String tableName, List<ColumnSchema> columns) {
+ super(tableName, columns);
}
- public abstract void open() throws Exception;
+ public String getTreeDBName() {
+ return "root.tree";
+ }
- public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
+ public Map<String, String> getMeasurementColumnNameMap() {
+ return Collections.emptyMap();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
index 90df96cc090..294a4d78a7b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/TableDeviceSchemaFetcher.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.queryengine.plan.Coordinator;
import org.apache.iotdb.db.queryengine.plan.execution.ExecutionResult;
import org.apache.iotdb.db.queryengine.plan.planner.LocalExecutionPlanner;
import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.schema.ConvertSchemaPredicateToFilterVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceSchemaCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AbstractTraverseDevice;
@@ -335,7 +336,7 @@ public class TableDeviceSchemaFetcher {
}
final DeviceEntry deviceEntry =
- new DeviceEntry(
+ new AlignedDeviceEntry(
deviceID,
attributeColumns.stream().map(attributeMap::get).collect(Collectors.toList()));
// TODO table metadata: process cases that selected attr columns different from those used for
@@ -423,7 +424,7 @@ public class TableDeviceSchemaFetcher {
attributeMap, nodes, table, columnHeaderList, columns, tableInstance, i);
final IDeviceID deviceID = IDeviceID.Factory.DEFAULT_FACTORY.create(nodes);
final DeviceEntry deviceEntry =
- new DeviceEntry(
+ new AlignedDeviceEntry(
deviceID,
attributeColumns.stream().map(attributeMap::get).collect(Collectors.toList()));
mppQueryContext.reserveMemoryForFrontEnd(deviceEntry.ramBytesUsed());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
index e825245f16e..5f08926fe70 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/RelationPlanner.java
@@ -40,6 +40,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Scope;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableMetadataImpl;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.TreeDeviceViewSchema;
import org.apache.iotdb.db.queryengine.plan.relational.planner.ir.IrUtils;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
@@ -47,6 +48,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationS
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AliasedRelation;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CoalesceExpression;
@@ -200,6 +202,24 @@ public class RelationPlanner extends AstVisitor<RelationPlan, Void> {
Map<Symbol, ColumnSchema> tableColumnSchema = symbolToColumnSchema.build();
analysis.addTableSchema(qualifiedObjectName, tableColumnSchema);
+
+ if (analysis.getTableHandle(table) instanceof TreeDeviceViewSchema) {
+ TreeDeviceViewSchema treeDeviceViewSchema =
+ (TreeDeviceViewSchema) analysis.getTableHandle(table);
+ return new RelationPlan(
+ new TreeDeviceViewScanNode(
+ idAllocator.genPlanNodeId(),
+ qualifiedObjectName,
+ outputSymbols,
+ tableColumnSchema,
+ idAndAttributeIndexMap,
+ treeDeviceViewSchema.getTreeDBName(),
+ treeDeviceViewSchema.getMeasurementColumnNameMap()),
+ scope,
+ outputSymbols,
+ outerContext);
+ }
+
TableScanNode tableScanNode =
qualifiedObjectName.getDatabaseName().equals(INFORMATION_DATABASE)
? new InformationSchemaTableScanNode(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
index 470d616f44d..53d6516e2e3 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/distribute/TableDistributedPlanGenerator.java
@@ -26,12 +26,14 @@ import org.apache.iotdb.commons.partition.SchemaPartition;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.planner.TableOperatorGenerator;
import org.apache.iotdb.db.queryengine.plan.planner.distribution.NodeDistribution;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.WritePlanNode;
import org.apache.iotdb.db.queryengine.plan.relational.analyzer.Analysis;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
@@ -40,6 +42,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
@@ -57,6 +60,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.AbstractTableDeviceQueryNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.schema.TableDeviceFetchNode;
@@ -91,6 +97,7 @@ import java.util.stream.IntStream;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static org.apache.iotdb.commons.partition.DataPartition.NOT_ASSIGNED;
import static org.apache.iotdb.commons.udf.builtin.relational.TableBuiltinScalarFunction.DATE_BIN;
+import static org.apache.iotdb.db.queryengine.plan.planner.TableOperatorGenerator.createTreeDeviceIdColumnValueExtractor;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator.GROUP_KEY_SUFFIX;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator.SEPARATOR;
import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.Step.SINGLE;
@@ -486,7 +493,8 @@ public class TableDistributedPlanGenerator
node.getPushDownPredicate(),
node.getPushDownLimit(),
node.getPushDownOffset(),
- node.isPushLimitToEachDevice());
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice());
scanNode.setRegionReplicaSet(regionReplicaSet);
return scanNode;
});
@@ -504,15 +512,12 @@ public class TableDistributedPlanGenerator
int maxDeviceEntrySizeOfTableScan = 0;
for (final Map.Entry<TRegionReplicaSet, DeviceTableScanNode> entry :
tableScanNodeMap.entrySet()) {
- final TRegionReplicaSet regionReplicaSet = entry.getKey();
final DeviceTableScanNode subDeviceTableScanNode = entry.getValue();
- subDeviceTableScanNode.setPlanNodeId(queryId.genPlanNodeId());
- subDeviceTableScanNode.setRegionReplicaSet(regionReplicaSet);
resultTableScanNodeList.add(subDeviceTableScanNode);
if (mostUsedDataRegion == null
|| subDeviceTableScanNode.getDeviceEntries().size() > maxDeviceEntrySizeOfTableScan) {
- mostUsedDataRegion = regionReplicaSet;
+ mostUsedDataRegion = entry.getKey();
maxDeviceEntrySizeOfTableScan = subDeviceTableScanNode.getDeviceEntries().size();
}
}
@@ -526,6 +531,119 @@ public class TableDistributedPlanGenerator
return resultTableScanNodeList;
}
+ @Override
+ public List<PlanNode> visitTreeDeviceViewScan(TreeDeviceViewScanNode node, PlanContext context) {
+ Map<TRegionReplicaSet, Pair<TreeAlignedDeviceViewScanNode, TreeNonAlignedDeviceViewScanNode>>
+ tableScanNodeMap = new HashMap<>();
+
+ for (DeviceEntry deviceEntry : node.getDeviceEntries()) {
+ List<TRegionReplicaSet> regionReplicaSets =
+ analysis.getDataRegionReplicaSetWithTimeFilter(
+ node.getTreeDBName(), deviceEntry.getDeviceID(), node.getTimeFilter());
+
+ for (TRegionReplicaSet regionReplicaSet : regionReplicaSets) {
+ boolean aligned = deviceEntry instanceof AlignedDeviceEntry;
+ Pair<TreeAlignedDeviceViewScanNode, TreeNonAlignedDeviceViewScanNode> pair =
+ tableScanNodeMap.get(regionReplicaSet);
+
+ if (pair == null) {
+ pair = new Pair<>(null, null);
+ tableScanNodeMap.put(regionReplicaSet, pair);
+ }
+
+ if (pair.left == null && aligned) {
+ TreeAlignedDeviceViewScanNode scanNode =
+ new TreeAlignedDeviceViewScanNode(
+ queryId.genPlanNodeId(),
+ node.getQualifiedObjectName(),
+ node.getOutputSymbols(),
+ node.getAssignments(),
+ new ArrayList<>(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice(),
+ node.getTreeDBName(),
+ node.getMeasurementColumnNameMap());
+ scanNode.setRegionReplicaSet(regionReplicaSet);
+ pair.left = scanNode;
+ }
+
+ if (pair.right == null && !aligned) {
+ TreeNonAlignedDeviceViewScanNode scanNode =
+ new TreeNonAlignedDeviceViewScanNode(
+ queryId.genPlanNodeId(),
+ node.getQualifiedObjectName(),
+ node.getOutputSymbols(),
+ node.getAssignments(),
+ new ArrayList<>(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice(),
+ node.getTreeDBName(),
+ node.getMeasurementColumnNameMap());
+ scanNode.setRegionReplicaSet(regionReplicaSet);
+ pair.right = scanNode;
+ }
+
+ if (aligned) {
+ pair.left.appendDeviceEntry(deviceEntry);
+ } else {
+ pair.right.appendDeviceEntry(deviceEntry);
+ }
+ }
+ }
+
+ if (tableScanNodeMap.isEmpty()) {
+ node.setRegionReplicaSet(NOT_ASSIGNED);
+ return Collections.singletonList(node);
+ }
+
+ List<PlanNode> resultTableScanNodeList = new ArrayList<>();
+ TRegionReplicaSet mostUsedDataRegion = null;
+ int maxDeviceEntrySizeOfTableScan = 0;
+ for (Map.Entry<
+ TRegionReplicaSet,
+ Pair<TreeAlignedDeviceViewScanNode, TreeNonAlignedDeviceViewScanNode>>
+ entry : tableScanNodeMap.entrySet()) {
+ TRegionReplicaSet regionReplicaSet = entry.getKey();
+ Pair<TreeAlignedDeviceViewScanNode, TreeNonAlignedDeviceViewScanNode> pair = entry.getValue();
+ int currentDeviceEntrySize = 0;
+
+ if (pair.left != null) {
+ currentDeviceEntrySize += pair.left.getDeviceEntries().size();
+ resultTableScanNodeList.add(pair.left);
+ }
+
+ if (pair.right != null) {
+ currentDeviceEntrySize += pair.right.getDeviceEntries().size();
+ resultTableScanNodeList.add(pair.right);
+ }
+
+ if (mostUsedDataRegion == null || currentDeviceEntrySize > maxDeviceEntrySizeOfTableScan) {
+ mostUsedDataRegion = regionReplicaSet;
+ maxDeviceEntrySizeOfTableScan = currentDeviceEntrySize;
+ }
+ }
+ context.mostUsedRegion = mostUsedDataRegion;
+
+ if (!context.hasSortProperty) {
+ return resultTableScanNodeList;
+ }
+
+ processSortProperty(node, resultTableScanNodeList, context);
+ return resultTableScanNodeList;
+ }
+
@Override
public List<PlanNode> visitInformationSchemaTableScan(
InformationSchemaTableScanNode node, PlanContext context) {
@@ -605,7 +723,9 @@ public class TableDistributedPlanGenerator
for (DeviceEntry deviceEntry : node.getDeviceEntries()) {
List<TRegionReplicaSet> regionReplicaSets =
analysis.getDataRegionReplicaSetWithTimeFilter(
- node.getQualifiedObjectName().getDatabaseName(),
+ node instanceof AggregationTreeDeviceViewScanNode
+ ? ((AggregationTreeDeviceViewScanNode) node).getTreeDBName()
+ : node.getQualifiedObjectName().getDatabaseName(),
deviceEntry.getDeviceID(),
node.getTimeFilter());
if (regionReplicaSets.size() > 1) {
@@ -646,15 +766,12 @@ public class TableDistributedPlanGenerator
TRegionReplicaSet mostUsedDataRegion = null;
int maxDeviceEntrySizeOfTableScan = 0;
for (Map.Entry<TRegionReplicaSet, AggregationTableScanNode> entry : regionNodeMap.entrySet()) {
- TRegionReplicaSet regionReplicaSet = entry.getKey();
DeviceTableScanNode subDeviceTableScanNode = entry.getValue();
- subDeviceTableScanNode.setPlanNodeId(queryId.genPlanNodeId());
- subDeviceTableScanNode.setRegionReplicaSet(regionReplicaSet);
resultTableScanNodeList.add(subDeviceTableScanNode);
if (mostUsedDataRegion == null
|| subDeviceTableScanNode.getDeviceEntries().size() > maxDeviceEntrySizeOfTableScan) {
- mostUsedDataRegion = regionReplicaSet;
+ mostUsedDataRegion = entry.getKey();
maxDeviceEntrySizeOfTableScan = subDeviceTableScanNode.getDeviceEntries().size();
}
}
@@ -698,6 +815,14 @@ public class TableDistributedPlanGenerator
List<List<TRegionReplicaSet>> regionReplicaSetsList,
Map<TRegionReplicaSet, AggregationTableScanNode> regionNodeMap,
AggregationTableScanNode partialAggTableScanNode) {
+ AggregationTreeDeviceViewScanNode aggregationTreeDeviceViewScanNode;
+ if (originalAggTableScanNode instanceof AggregationTreeDeviceViewScanNode) {
+ aggregationTreeDeviceViewScanNode =
+ (AggregationTreeDeviceViewScanNode) originalAggTableScanNode;
+ } else {
+ aggregationTreeDeviceViewScanNode = null;
+ }
+
for (int i = 0; i < regionReplicaSetsList.size(); i++) {
for (TRegionReplicaSet regionReplicaSet : regionReplicaSetsList.get(i)) {
AggregationTableScanNode aggregationTableScanNode =
@@ -705,25 +830,49 @@ public class TableDistributedPlanGenerator
regionReplicaSet,
k -> {
AggregationTableScanNode scanNode =
- new AggregationTableScanNode(
- queryId.genPlanNodeId(),
- partialAggTableScanNode.getQualifiedObjectName(),
- partialAggTableScanNode.getOutputSymbols(),
- partialAggTableScanNode.getAssignments(),
- new ArrayList<>(),
- partialAggTableScanNode.getIdAndAttributeIndexMap(),
- partialAggTableScanNode.getScanOrder(),
- partialAggTableScanNode.getTimePredicate().orElse(null),
- partialAggTableScanNode.getPushDownPredicate(),
- partialAggTableScanNode.getPushDownLimit(),
- partialAggTableScanNode.getPushDownOffset(),
- partialAggTableScanNode.isPushLimitToEachDevice(),
- partialAggTableScanNode.getProjection(),
- partialAggTableScanNode.getAggregations(),
- partialAggTableScanNode.getGroupingSets(),
- partialAggTableScanNode.getPreGroupedSymbols(),
- partialAggTableScanNode.getStep(),
- partialAggTableScanNode.getGroupIdSymbol());
+ (aggregationTreeDeviceViewScanNode == null)
+ ? new AggregationTableScanNode(
+ queryId.genPlanNodeId(),
+ partialAggTableScanNode.getQualifiedObjectName(),
+ partialAggTableScanNode.getOutputSymbols(),
+ partialAggTableScanNode.getAssignments(),
+ new ArrayList<>(),
+ partialAggTableScanNode.getIdAndAttributeIndexMap(),
+ partialAggTableScanNode.getScanOrder(),
+ partialAggTableScanNode.getTimePredicate().orElse(null),
+ partialAggTableScanNode.getPushDownPredicate(),
+ partialAggTableScanNode.getPushDownLimit(),
+ partialAggTableScanNode.getPushDownOffset(),
+ partialAggTableScanNode.isPushLimitToEachDevice(),
+ partialAggTableScanNode.containsNonAlignedDevice(),
+ partialAggTableScanNode.getProjection(),
+ partialAggTableScanNode.getAggregations(),
+ partialAggTableScanNode.getGroupingSets(),
+ partialAggTableScanNode.getPreGroupedSymbols(),
+ partialAggTableScanNode.getStep(),
+ partialAggTableScanNode.getGroupIdSymbol())
+ : new AggregationTreeDeviceViewScanNode(
+ queryId.genPlanNodeId(),
+ partialAggTableScanNode.getQualifiedObjectName(),
+ partialAggTableScanNode.getOutputSymbols(),
+ partialAggTableScanNode.getAssignments(),
+ new ArrayList<>(),
+ partialAggTableScanNode.getIdAndAttributeIndexMap(),
+ partialAggTableScanNode.getScanOrder(),
+ partialAggTableScanNode.getTimePredicate().orElse(null),
+ partialAggTableScanNode.getPushDownPredicate(),
+ partialAggTableScanNode.getPushDownLimit(),
+ partialAggTableScanNode.getPushDownOffset(),
+ partialAggTableScanNode.isPushLimitToEachDevice(),
+ partialAggTableScanNode.containsNonAlignedDevice(),
+ partialAggTableScanNode.getProjection(),
+ partialAggTableScanNode.getAggregations(),
+ partialAggTableScanNode.getGroupingSets(),
+ partialAggTableScanNode.getPreGroupedSymbols(),
+ partialAggTableScanNode.getStep(),
+ partialAggTableScanNode.getGroupIdSymbol(),
+ aggregationTreeDeviceViewScanNode.getTreeDBName(),
+ aggregationTreeDeviceViewScanNode.getMeasurementColumnNameMap());
scanNode.setRegionReplicaSet(regionReplicaSet);
return scanNode;
});
@@ -801,6 +950,8 @@ public class TableDistributedPlanGenerator
return;
}
+ Optional<IDeviceID.TreeDeviceIdColumnValueExtractor> extractor =
+ createTreeDeviceIdColumnValueExtractor(deviceTableScanNode);
final List<Function<DeviceEntry, String>> orderingRules = new ArrayList<>();
for (final Symbol symbol : newOrderingSymbols) {
final Integer idx = deviceTableScanNode.getIdAndAttributeIndexMap().get(symbol);
@@ -810,8 +961,18 @@ public class TableDistributedPlanGenerator
}
if (deviceTableScanNode.getAssignments().get(symbol).getColumnCategory()
== TsTableColumnCategory.TAG) {
- // segments[0] is always tableName
- orderingRules.add(deviceEntry -> (String) deviceEntry.getNthSegment(idx + 1));
+
+ // segments[0] is always tableName for table model
+ Function<DeviceEntry, String> iDColumnFunction =
+ extractor
+ .<Function<DeviceEntry, String>>map(
+ treeDeviceIdColumnValueExtractor ->
+ deviceEntry ->
+ (String)
+ treeDeviceIdColumnValueExtractor.extract(
+ deviceEntry.getDeviceID(), idx))
+ .orElseGet(() -> deviceEntry -> (String) deviceEntry.getNthSegment(idx + 1));
+ orderingRules.add(iDColumnFunction);
} else {
orderingRules.add(
deviceEntry ->
@@ -883,6 +1044,21 @@ public class TableDistributedPlanGenerator
}
}
+ private Optional<IDeviceID.TreeDeviceIdColumnValueExtractor>
+ createTreeDeviceIdColumnValueExtractor(DeviceTableScanNode node) {
+ if (node instanceof TreeDeviceViewScanNode) {
+ return Optional.of(
+ TableOperatorGenerator.createTreeDeviceIdColumnValueExtractor(
+ ((TreeDeviceViewScanNode) node).getTreeDBName()));
+ } else if (node instanceof AggregationTreeDeviceViewScanNode) {
+ return Optional.of(
+ TableOperatorGenerator.createTreeDeviceIdColumnValueExtractor(
+ ((AggregationTreeDeviceViewScanNode) node).getTreeDBName()));
+ } else {
+ return Optional.empty();
+ }
+ }
+
private Optional<OrderingScheme> tableScanOrderingSchema(
Map<Symbol, ColumnSchema> tableColumnSchema,
List<Symbol> newOrderingSymbols,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
index c9c577300e6..f311c5c2709 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/iterative/rule/PruneTableScanColumns.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationT
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.InformationSchemaTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import java.util.ArrayList;
import java.util.LinkedHashMap;
@@ -87,20 +88,43 @@ public class PruneTableScanColumns extends ProjectOffPushDownRule<TableScanNode>
.forEach(
symbol -> newAssignments.put(symbol, node.getAssignments().get(symbol))));
- return Optional.of(
- new DeviceTableScanNode(
- deviceTableScanNode.getPlanNodeId(),
- deviceTableScanNode.getQualifiedObjectName(),
- newOutputs,
- newAssignments,
- deviceTableScanNode.getDeviceEntries(),
- deviceTableScanNode.getIdAndAttributeIndexMap(),
- deviceTableScanNode.getScanOrder(),
- deviceTableScanNode.getTimePredicate().orElse(null),
- deviceTableScanNode.getPushDownPredicate(),
- deviceTableScanNode.getPushDownLimit(),
- deviceTableScanNode.getPushDownOffset(),
- deviceTableScanNode.isPushLimitToEachDevice()));
+ if (node instanceof TreeDeviceViewScanNode) {
+ TreeDeviceViewScanNode treeDeviceViewScanNode =
+ (TreeDeviceViewScanNode) deviceTableScanNode;
+ return Optional.of(
+ new TreeDeviceViewScanNode(
+ deviceTableScanNode.getPlanNodeId(),
+ deviceTableScanNode.getQualifiedObjectName(),
+ newOutputs,
+ newAssignments,
+ deviceTableScanNode.getDeviceEntries(),
+ deviceTableScanNode.getIdAndAttributeIndexMap(),
+ deviceTableScanNode.getScanOrder(),
+ deviceTableScanNode.getTimePredicate().orElse(null),
+ deviceTableScanNode.getPushDownPredicate(),
+ deviceTableScanNode.getPushDownLimit(),
+ deviceTableScanNode.getPushDownOffset(),
+ deviceTableScanNode.isPushLimitToEachDevice(),
+ deviceTableScanNode.containsNonAlignedDevice(),
+ treeDeviceViewScanNode.getTreeDBName(),
+ treeDeviceViewScanNode.getMeasurementColumnNameMap()));
+ } else {
+ return Optional.of(
+ new DeviceTableScanNode(
+ deviceTableScanNode.getPlanNodeId(),
+ deviceTableScanNode.getQualifiedObjectName(),
+ newOutputs,
+ newAssignments,
+ deviceTableScanNode.getDeviceEntries(),
+ deviceTableScanNode.getIdAndAttributeIndexMap(),
+ deviceTableScanNode.getScanOrder(),
+ deviceTableScanNode.getTimePredicate().orElse(null),
+ deviceTableScanNode.getPushDownPredicate(),
+ deviceTableScanNode.getPushDownLimit(),
+ deviceTableScanNode.getPushDownOffset(),
+ deviceTableScanNode.isPushLimitToEachDevice(),
+ deviceTableScanNode.containsNonAlignedDevice()));
+ }
} else if (node instanceof InformationSchemaTableScanNode) {
// For the convenience of process in execution stage, column-prune for
// InformationSchemaTableScanNode is
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
index 01d58362fb1..f6628bc2ca8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTableScanNode.java
@@ -35,6 +35,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SymbolReference;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import org.apache.tsfile.read.common.type.LongType;
import org.apache.tsfile.read.common.type.TimestampType;
import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -44,7 +45,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -59,13 +59,13 @@ import static org.apache.iotdb.db.utils.constant.SqlConstant.TABLE_TIME_COLUMN_N
public class AggregationTableScanNode extends DeviceTableScanNode {
// if there is date_bin function of time, we should use this field to transform time input
- private final Assignments projection;
+ protected Assignments projection;
- private final Map<Symbol, AggregationNode.Aggregation> aggregations;
- private final AggregationNode.GroupingSetDescriptor groupingSets;
- private final List<Symbol> preGroupedSymbols;
- private AggregationNode.Step step;
- private final Optional<Symbol> groupIdSymbol;
+ protected Map<Symbol, AggregationNode.Aggregation> aggregations;
+ protected AggregationNode.GroupingSetDescriptor groupingSets;
+ protected List<Symbol> preGroupedSymbols;
+ protected AggregationNode.Step step;
+ protected Optional<Symbol> groupIdSymbol;
public AggregationTableScanNode(
PlanNodeId id,
@@ -80,6 +80,7 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
long pushDownLimit,
long pushDownOffset,
boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice,
Assignments projection,
Map<Symbol, AggregationNode.Aggregation> aggregations,
AggregationNode.GroupingSetDescriptor groupingSets,
@@ -98,30 +99,11 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
pushDownPredicate,
pushDownLimit,
pushDownOffset,
- pushLimitToEachDevice);
+ pushLimitToEachDevice,
+ containsNonAlignedDevice);
this.projection = projection;
- this.aggregations = new LinkedHashMap<>(aggregations.size());
- for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : aggregations.entrySet()) {
- Symbol symbol = entry.getKey();
- AggregationNode.Aggregation aggregation = entry.getValue();
- if (aggregation.getArguments().isEmpty()
- && COUNT.equals(aggregation.getResolvedFunction().getSignature().getName())) {
- AggregationNode.Aggregation countStarAggregation = getCountStarAggregation(aggregation);
- if (!assignments.containsKey(Symbol.of(TABLE_TIME_COLUMN_NAME))) {
- assignments.put(
- Symbol.of(TABLE_TIME_COLUMN_NAME),
- new ColumnSchema(
- TABLE_TIME_COLUMN_NAME,
- TimestampType.TIMESTAMP,
- false,
- TsTableColumnCategory.TIME));
- }
- this.aggregations.put(symbol, countStarAggregation);
- } else {
- this.aggregations.put(symbol, aggregation);
- }
- }
+ this.aggregations = transformCountStar(aggregations, assignments);
aggregations.values().forEach(aggregation -> aggregation.verifyArguments(step));
requireNonNull(groupingSets, "groupingSets is null");
@@ -151,14 +133,51 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
"Pre-grouped symbols must be a subset of the grouping keys");
this.preGroupedSymbols = ImmutableList.copyOf(preGroupedSymbols);
+ this.setOutputSymbols(constructOutputSymbols(groupingSets, aggregations));
+ }
+
+ protected AggregationTableScanNode() {}
+
+ private static List<Symbol> constructOutputSymbols(
+ AggregationNode.GroupingSetDescriptor groupingSets,
+ Map<Symbol, AggregationNode.Aggregation> aggregations) {
ImmutableList.Builder<Symbol> outputs = ImmutableList.builder();
outputs.addAll(groupingSets.getGroupingKeys());
outputs.addAll(aggregations.keySet());
+ return outputs.build();
+ }
+
+ // transform count() to count(time)
+ private static Map<Symbol, AggregationNode.Aggregation> transformCountStar(
+ Map<Symbol, AggregationNode.Aggregation> aggregations,
+ Map<Symbol, ColumnSchema> assignments) {
+ ImmutableMap.Builder<Symbol, AggregationNode.Aggregation> resultBuilder =
+ ImmutableMap.builder();
- this.setOutputSymbols(outputs.build());
+ for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : aggregations.entrySet()) {
+ Symbol symbol = entry.getKey();
+ AggregationNode.Aggregation aggregation = entry.getValue();
+ if (aggregation.getArguments().isEmpty()) {
+ AggregationNode.Aggregation countStarAggregation = getCountStarAggregation(aggregation);
+ if (!getTimeColumn(assignments).isPresent()) {
+ assignments.put(
+ Symbol.of(TABLE_TIME_COLUMN_NAME),
+ new ColumnSchema(
+ TABLE_TIME_COLUMN_NAME,
+ TimestampType.TIMESTAMP,
+ false,
+ TsTableColumnCategory.TIME));
+ }
+ resultBuilder.put(symbol, countStarAggregation);
+ } else {
+ resultBuilder.put(symbol, aggregation);
+ }
+ }
+
+ return resultBuilder.build();
}
- private AggregationNode.Aggregation getCountStarAggregation(
+ private static AggregationNode.Aggregation getCountStarAggregation(
AggregationNode.Aggregation aggregation) {
ResolvedFunction resolvedFunction = aggregation.getResolvedFunction();
ResolvedFunction countStarFunction =
@@ -237,6 +256,7 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
pushDownLimit,
pushDownOffset,
pushLimitToEachDevice,
+ containsNonAlignedDevice,
projection,
aggregations,
groupingSets,
@@ -255,6 +275,32 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
AggregationNode aggregationNode,
ProjectNode projectNode,
DeviceTableScanNode tableScanNode) {
+ if (tableScanNode instanceof TreeDeviceViewScanNode) {
+ TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) tableScanNode;
+ return new AggregationTreeDeviceViewScanNode(
+ id,
+ tableScanNode.getQualifiedObjectName(),
+ tableScanNode.getOutputSymbols(),
+ tableScanNode.getAssignments(),
+ tableScanNode.getDeviceEntries(),
+ tableScanNode.getIdAndAttributeIndexMap(),
+ tableScanNode.getScanOrder(),
+ tableScanNode.getTimePredicate().orElse(null),
+ tableScanNode.getPushDownPredicate(),
+ tableScanNode.getPushDownLimit(),
+ tableScanNode.getPushDownOffset(),
+ tableScanNode.isPushLimitToEachDevice(),
+ tableScanNode.containsNonAlignedDevice(),
+ projectNode == null ? null : projectNode.getAssignments(),
+ aggregationNode.getAggregations(),
+ aggregationNode.getGroupingSets(),
+ aggregationNode.getPreGroupedSymbols(),
+ aggregationNode.getStep(),
+ aggregationNode.getGroupIdSymbol(),
+ treeDeviceViewScanNode.getTreeDBName(),
+ treeDeviceViewScanNode.getMeasurementColumnNameMap());
+ }
+
return new AggregationTableScanNode(
id,
tableScanNode.getQualifiedObjectName(),
@@ -268,6 +314,7 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
tableScanNode.getPushDownLimit(),
tableScanNode.getPushDownOffset(),
tableScanNode.isPushLimitToEachDevice(),
+ tableScanNode.containsNonAlignedDevice(),
projectNode == null ? null : projectNode.getAssignments(),
aggregationNode.getAggregations(),
aggregationNode.getGroupingSets(),
@@ -282,6 +329,32 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
ProjectNode projectNode,
DeviceTableScanNode tableScanNode,
AggregationNode.Step step) {
+ if (tableScanNode instanceof TreeDeviceViewScanNode) {
+ TreeDeviceViewScanNode treeDeviceViewScanNode = (TreeDeviceViewScanNode) tableScanNode;
+ return new AggregationTreeDeviceViewScanNode(
+ id,
+ tableScanNode.getQualifiedObjectName(),
+ tableScanNode.getOutputSymbols(),
+ tableScanNode.getAssignments(),
+ tableScanNode.getDeviceEntries(),
+ tableScanNode.getIdAndAttributeIndexMap(),
+ tableScanNode.getScanOrder(),
+ tableScanNode.getTimePredicate().orElse(null),
+ tableScanNode.getPushDownPredicate(),
+ tableScanNode.getPushDownLimit(),
+ tableScanNode.getPushDownOffset(),
+ tableScanNode.isPushLimitToEachDevice(),
+ tableScanNode.containsNonAlignedDevice(),
+ projectNode == null ? null : projectNode.getAssignments(),
+ aggregationNode.getAggregations(),
+ aggregationNode.getGroupingSets(),
+ aggregationNode.getPreGroupedSymbols(),
+ step,
+ aggregationNode.getGroupIdSymbol(),
+ treeDeviceViewScanNode.getTreeDBName(),
+ treeDeviceViewScanNode.getMeasurementColumnNameMap());
+ }
+
return new AggregationTableScanNode(
id,
tableScanNode.getQualifiedObjectName(),
@@ -295,6 +368,7 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
tableScanNode.getPushDownLimit(),
tableScanNode.getPushDownOffset(),
tableScanNode.isPushLimitToEachDevice(),
+ tableScanNode.containsNonAlignedDevice(),
projectNode == null ? null : projectNode.getAssignments(),
aggregationNode.getAggregations(),
aggregationNode.getGroupingSets(),
@@ -336,59 +410,14 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
return "AggregationTableScanNode-" + this.getPlanNodeId();
}
- @Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {
- PlanNodeType.TABLE_AGGREGATION_TABLE_SCAN_NODE.serialize(byteBuffer);
+ protected static void serializeMemberVariables(
+ AggregationTableScanNode node, ByteBuffer byteBuffer) {
+ DeviceTableScanNode.serializeMemberVariables(node, byteBuffer, false);
- if (qualifiedObjectName.getDatabaseName() != null) {
+ if (node.projection != null) {
ReadWriteIOUtils.write(true, byteBuffer);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), byteBuffer);
-
- ReadWriteIOUtils.write(assignments.size(), byteBuffer);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), byteBuffer);
- ColumnSchema.serialize(entry.getValue(), byteBuffer);
- }
-
- ReadWriteIOUtils.write(deviceEntries.size(), byteBuffer);
- for (DeviceEntry entry : deviceEntries) {
- entry.serialize(byteBuffer);
- }
-
- ReadWriteIOUtils.write(idAndAttributeIndexMap.size(), byteBuffer);
- for (Map.Entry<Symbol, Integer> entry : idAndAttributeIndexMap.entrySet()) {
- Symbol.serialize(entry.getKey(), byteBuffer);
- ReadWriteIOUtils.write(entry.getValue(), byteBuffer);
- }
-
- ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
-
- if (timePredicate != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- Expression.serialize(timePredicate, byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
-
- if (pushDownPredicate != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- Expression.serialize(pushDownPredicate, byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
-
- ReadWriteIOUtils.write(pushDownLimit, byteBuffer);
- ReadWriteIOUtils.write(pushDownOffset, byteBuffer);
- ReadWriteIOUtils.write(pushLimitToEachDevice, byteBuffer);
-
- if (projection != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- ReadWriteIOUtils.write(projection.getMap().size(), byteBuffer);
- for (Map.Entry<Symbol, Expression> entry : projection.getMap().entrySet()) {
+ ReadWriteIOUtils.write(node.projection.getMap().size(), byteBuffer);
+ for (Map.Entry<Symbol, Expression> entry : node.projection.getMap().entrySet()) {
Symbol.serialize(entry.getKey(), byteBuffer);
Expression.serialize(entry.getValue(), byteBuffer);
}
@@ -396,79 +425,36 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
ReadWriteIOUtils.write(false, byteBuffer);
}
- ReadWriteIOUtils.write(aggregations.size(), byteBuffer);
- for (Map.Entry<Symbol, AggregationNode.Aggregation> aggregation : aggregations.entrySet()) {
+ ReadWriteIOUtils.write(node.aggregations.size(), byteBuffer);
+ for (Map.Entry<Symbol, AggregationNode.Aggregation> aggregation :
+ node.aggregations.entrySet()) {
Symbol.serialize(aggregation.getKey(), byteBuffer);
aggregation.getValue().serialize(byteBuffer);
}
- groupingSets.serialize(byteBuffer);
+ node.groupingSets.serialize(byteBuffer);
- ReadWriteIOUtils.write(preGroupedSymbols.size(), byteBuffer);
- for (Symbol preGroupedSymbol : preGroupedSymbols) {
+ ReadWriteIOUtils.write(node.preGroupedSymbols.size(), byteBuffer);
+ for (Symbol preGroupedSymbol : node.preGroupedSymbols) {
Symbol.serialize(preGroupedSymbol, byteBuffer);
}
- step.serialize(byteBuffer);
+ node.step.serialize(byteBuffer);
- ReadWriteIOUtils.write(groupIdSymbol.isPresent(), byteBuffer);
- if (groupIdSymbol.isPresent()) {
- Symbol.serialize(groupIdSymbol.get(), byteBuffer);
+ ReadWriteIOUtils.write(node.groupIdSymbol.isPresent(), byteBuffer);
+ if (node.groupIdSymbol.isPresent()) {
+ Symbol.serialize(node.groupIdSymbol.get(), byteBuffer);
}
}
- @Override
- protected void serializeAttributes(DataOutputStream stream) throws IOException {
- PlanNodeType.TABLE_AGGREGATION_TABLE_SCAN_NODE.serialize(stream);
- if (qualifiedObjectName.getDatabaseName() != null) {
- ReadWriteIOUtils.write(true, stream);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), stream);
-
- ReadWriteIOUtils.write(assignments.size(), stream);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), stream);
- ColumnSchema.serialize(entry.getValue(), stream);
- }
-
- ReadWriteIOUtils.write(deviceEntries.size(), stream);
- for (DeviceEntry entry : deviceEntries) {
- entry.serialize(stream);
- }
-
- ReadWriteIOUtils.write(idAndAttributeIndexMap.size(), stream);
- for (Map.Entry<Symbol, Integer> entry : idAndAttributeIndexMap.entrySet()) {
- Symbol.serialize(entry.getKey(), stream);
- ReadWriteIOUtils.write(entry.getValue(), stream);
- }
-
- ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
-
- if (timePredicate != null) {
- ReadWriteIOUtils.write(true, stream);
- Expression.serialize(timePredicate, stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
-
- if (pushDownPredicate != null) {
- ReadWriteIOUtils.write(true, stream);
- Expression.serialize(pushDownPredicate, stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
-
- ReadWriteIOUtils.write(pushDownLimit, stream);
- ReadWriteIOUtils.write(pushDownOffset, stream);
- ReadWriteIOUtils.write(pushLimitToEachDevice, stream);
+ protected static void serializeMemberVariables(
+ AggregationTableScanNode node, DataOutputStream stream) throws IOException {
+ DeviceTableScanNode.serializeMemberVariables(node, stream, false);
- if (projection != null) {
+ if (node.projection != null) {
ReadWriteIOUtils.write(true, stream);
- ReadWriteIOUtils.write(projection.getMap().size(), stream);
- for (Map.Entry<Symbol, Expression> entry : projection.getMap().entrySet()) {
+ ReadWriteIOUtils.write(node.projection.getMap().size(), stream);
+ for (Map.Entry<Symbol, Expression> entry : node.projection.getMap().entrySet()) {
Symbol.serialize(entry.getKey(), stream);
Expression.serialize(entry.getValue(), stream);
}
@@ -476,72 +462,33 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
ReadWriteIOUtils.write(false, stream);
}
- ReadWriteIOUtils.write(aggregations.size(), stream);
- for (Map.Entry<Symbol, AggregationNode.Aggregation> aggregation : aggregations.entrySet()) {
+ ReadWriteIOUtils.write(node.aggregations.size(), stream);
+ for (Map.Entry<Symbol, AggregationNode.Aggregation> aggregation :
+ node.aggregations.entrySet()) {
Symbol.serialize(aggregation.getKey(), stream);
aggregation.getValue().serialize(stream);
}
- groupingSets.serialize(stream);
+ node.groupingSets.serialize(stream);
- ReadWriteIOUtils.write(preGroupedSymbols.size(), stream);
- for (Symbol preGroupedSymbol : preGroupedSymbols) {
+ ReadWriteIOUtils.write(node.preGroupedSymbols.size(), stream);
+ for (Symbol preGroupedSymbol : node.preGroupedSymbols) {
Symbol.serialize(preGroupedSymbol, stream);
}
- step.serialize(stream);
+ node.step.serialize(stream);
- ReadWriteIOUtils.write(groupIdSymbol.isPresent(), stream);
- if (groupIdSymbol.isPresent()) {
- Symbol.serialize(groupIdSymbol.get(), stream);
+ ReadWriteIOUtils.write(node.groupIdSymbol.isPresent(), stream);
+ if (node.groupIdSymbol.isPresent()) {
+ Symbol.serialize(node.groupIdSymbol.get(), stream);
}
}
- public static AggregationTableScanNode deserialize(ByteBuffer byteBuffer) {
- boolean hasDatabaseName = ReadWriteIOUtils.readBool(byteBuffer);
- String databaseName = null;
- if (hasDatabaseName) {
- databaseName = ReadWriteIOUtils.readString(byteBuffer);
- }
- String tableName = ReadWriteIOUtils.readString(byteBuffer);
- QualifiedObjectName qualifiedObjectName = new QualifiedObjectName(databaseName, tableName);
-
- int size = ReadWriteIOUtils.readInt(byteBuffer);
- Map<Symbol, ColumnSchema> assignments = new LinkedHashMap<>(size);
- for (int i = 0; i < size; i++) {
- assignments.put(Symbol.deserialize(byteBuffer), ColumnSchema.deserialize(byteBuffer));
- }
-
- size = ReadWriteIOUtils.readInt(byteBuffer);
- List<DeviceEntry> deviceEntries = new ArrayList<>(size);
- while (size-- > 0) {
- deviceEntries.add(DeviceEntry.deserialize(byteBuffer));
- }
+ protected static void deserializeMemberVariables(
+ ByteBuffer byteBuffer, AggregationTableScanNode node) {
+ DeviceTableScanNode.deserializeMemberVariables(byteBuffer, node, false);
- size = ReadWriteIOUtils.readInt(byteBuffer);
- Map<Symbol, Integer> idAndAttributeIndexMap = new HashMap<>(size);
- while (size-- > 0) {
- idAndAttributeIndexMap.put(
- Symbol.deserialize(byteBuffer), ReadWriteIOUtils.readInt(byteBuffer));
- }
-
- Ordering scanOrder = Ordering.values()[ReadWriteIOUtils.readInt(byteBuffer)];
-
- Expression timePredicate = null;
- boolean hasTimePredicate = ReadWriteIOUtils.readBool(byteBuffer);
- if (hasTimePredicate) {
- timePredicate = Expression.deserialize(byteBuffer);
- }
-
- Expression pushDownPredicate = null;
- boolean hasPushDownPredicate = ReadWriteIOUtils.readBool(byteBuffer);
- if (hasPushDownPredicate) {
- pushDownPredicate = Expression.deserialize(byteBuffer);
- }
-
- long pushDownLimit = ReadWriteIOUtils.readLong(byteBuffer);
- long pushDownOffset = ReadWriteIOUtils.readLong(byteBuffer);
- boolean pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer);
+ int size;
Assignments.Builder projection = Assignments.builder();
boolean hasProjection = ReadWriteIOUtils.readBool(byteBuffer);
@@ -551,6 +498,7 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
projection.put(Symbol.deserialize(byteBuffer), Expression.deserialize(byteBuffer));
}
}
+ node.projection = projection.build();
size = ReadWriteIOUtils.readInt(byteBuffer);
final Map<Symbol, AggregationNode.Aggregation> aggregations = new LinkedHashMap<>(size);
@@ -559,42 +507,47 @@ public class AggregationTableScanNode extends DeviceTableScanNode {
Symbol.deserialize(byteBuffer), AggregationNode.Aggregation.deserialize(byteBuffer));
}
- AggregationNode.GroupingSetDescriptor groupingSetDescriptor =
- AggregationNode.GroupingSetDescriptor.deserialize(byteBuffer);
+ node.aggregations = transformCountStar(aggregations, node.getAssignments());
+
+ node.groupingSets = AggregationNode.GroupingSetDescriptor.deserialize(byteBuffer);
size = ReadWriteIOUtils.readInt(byteBuffer);
List<Symbol> preGroupedSymbols = new ArrayList<>(size);
while (size-- > 0) {
preGroupedSymbols.add(Symbol.deserialize(byteBuffer));
}
+ node.preGroupedSymbols = preGroupedSymbols;
- AggregationNode.Step step = AggregationNode.Step.deserialize(byteBuffer);
+ node.step = AggregationNode.Step.deserialize(byteBuffer);
Optional<Symbol> groupIdSymbol = Optional.empty();
if (ReadWriteIOUtils.readBool(byteBuffer)) {
groupIdSymbol = Optional.of(Symbol.deserialize(byteBuffer));
}
+ node.groupIdSymbol = groupIdSymbol;
- PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ node.outputSymbols = constructOutputSymbols(node.getGroupingSets(), node.getAggregations());
+ }
- return new AggregationTableScanNode(
- planNodeId,
- qualifiedObjectName,
- null,
- assignments,
- deviceEntries,
- idAndAttributeIndexMap,
- scanOrder,
- timePredicate,
- pushDownPredicate,
- pushDownLimit,
- pushDownOffset,
- pushLimitToEachDevice,
- projection.build(),
- aggregations,
- groupingSetDescriptor,
- preGroupedSymbols,
- step,
- groupIdSymbol);
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TABLE_AGGREGATION_TABLE_SCAN_NODE.serialize(byteBuffer);
+
+ AggregationTableScanNode.serializeMemberVariables(this, byteBuffer);
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TABLE_AGGREGATION_TABLE_SCAN_NODE.serialize(stream);
+
+ AggregationTableScanNode.serializeMemberVariables(this, stream);
+ }
+
+ public static AggregationTableScanNode deserialize(ByteBuffer byteBuffer) {
+ AggregationTableScanNode node = new AggregationTableScanNode();
+ AggregationTableScanNode.deserializeMemberVariables(byteBuffer, node);
+
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java
new file mode 100644
index 00000000000..9246e50a22b
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/AggregationTreeDeviceViewScanNode.java
@@ -0,0 +1,205 @@
+/*
+ * 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.plan.relational.planner.node;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+
+public class AggregationTreeDeviceViewScanNode extends AggregationTableScanNode {
+ private String treeDBName;
+ private Map<String, String> measurementColumnNameMap;
+
+ public AggregationTreeDeviceViewScanNode(
+ PlanNodeId id,
+ QualifiedObjectName qualifiedObjectName,
+ List<Symbol> outputSymbols,
+ Map<Symbol, ColumnSchema> assignments,
+ List<DeviceEntry> deviceEntries,
+ Map<Symbol, Integer> idAndAttributeIndexMap,
+ Ordering scanOrder,
+ Expression timePredicate,
+ Expression pushDownPredicate,
+ long pushDownLimit,
+ long pushDownOffset,
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice,
+ Assignments projection,
+ Map<Symbol, AggregationNode.Aggregation> aggregations,
+ AggregationNode.GroupingSetDescriptor groupingSets,
+ List<Symbol> preGroupedSymbols,
+ AggregationNode.Step step,
+ Optional<Symbol> groupIdSymbol,
+ String treeDBName,
+ Map<String, String> measurementColumnNameMap) {
+ super(
+ id,
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ projection,
+ aggregations,
+ groupingSets,
+ preGroupedSymbols,
+ step,
+ groupIdSymbol);
+ this.treeDBName = treeDBName;
+ this.measurementColumnNameMap = measurementColumnNameMap;
+ }
+
+ private AggregationTreeDeviceViewScanNode() {}
+
+ public String getTreeDBName() {
+ return treeDBName;
+ }
+
+ public Map<String, String> getMeasurementColumnNameMap() {
+ return measurementColumnNameMap;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitAggregationTreeDeviceViewScan(this, context);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ AggregationTreeDeviceViewScanNode that = (AggregationTreeDeviceViewScanNode) o;
+ return Objects.equals(treeDBName, that.treeDBName)
+ && Objects.equals(measurementColumnNameMap, that.measurementColumnNameMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), treeDBName, measurementColumnNameMap);
+ }
+
+ @Override
+ public String toString() {
+ return "AggregationTreeDeviceViewTableScanNode-" + this.getPlanNodeId();
+ }
+
+ @Override
+ public AggregationTreeDeviceViewScanNode clone() {
+ return new AggregationTreeDeviceViewScanNode(
+ id,
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ projection,
+ aggregations,
+ groupingSets,
+ preGroupedSymbols,
+ step,
+ groupIdSymbol,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.AGGREGATION_TREE_DEVICE_VIEW_SCAN_NODE.serialize(byteBuffer);
+
+ AggregationTableScanNode.serializeMemberVariables(this, byteBuffer);
+
+ ReadWriteIOUtils.write(treeDBName, byteBuffer);
+ ReadWriteIOUtils.write(measurementColumnNameMap.size(), byteBuffer);
+ for (Map.Entry<String, String> entry : measurementColumnNameMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue(), byteBuffer);
+ }
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.AGGREGATION_TREE_DEVICE_VIEW_SCAN_NODE.serialize(stream);
+
+ AggregationTableScanNode.serializeMemberVariables(this, stream);
+
+ ReadWriteIOUtils.write(treeDBName, stream);
+ ReadWriteIOUtils.write(measurementColumnNameMap.size(), stream);
+ for (Map.Entry<String, String> entry : measurementColumnNameMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), stream);
+ ReadWriteIOUtils.write(entry.getValue(), stream);
+ }
+ }
+
+ public static AggregationTreeDeviceViewScanNode deserialize(ByteBuffer byteBuffer) {
+ AggregationTreeDeviceViewScanNode node = new AggregationTreeDeviceViewScanNode();
+ AggregationTableScanNode.deserializeMemberVariables(byteBuffer, node, false);
+
+ node.treeDBName = ReadWriteIOUtils.readString(byteBuffer);
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<String, String> measurementColumnNameMap = new HashMap<>(size);
+ for (int i = 0; i < size; i++) {
+ measurementColumnNameMap.put(
+ ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
+ }
+ node.measurementColumnNameMap = measurementColumnNameMap;
+
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java
index 0e6e76b1e7b..e8a6d18ebc0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/DeviceTableScanNode.java
@@ -51,7 +51,7 @@ public class DeviceTableScanNode extends TableScanNode {
// For example, for DeviceEntry `table1.tag1.tag2.attribute1.attribute2.s1.s2`, the content of
// `idAndAttributeIndexMap` will
// be `tag1: 0, tag2: 1, attribute1: 0, attribute2: 1`.
- protected final Map<Symbol, Integer> idAndAttributeIndexMap;
+ protected Map<Symbol, Integer> idAndAttributeIndexMap;
// The order to traverse the data.
// Currently, we only support TIMESTAMP_ASC and TIMESTAMP_DESC here.
@@ -65,7 +65,7 @@ public class DeviceTableScanNode extends TableScanNode {
// 1 or time < 10
@Nullable protected Expression timePredicate;
- protected Filter timeFilter;
+ protected transient Filter timeFilter;
// pushLimitToEachDevice == true means that each device in DeviceTableScanNode need to return
// `pushDownLimit` row number
@@ -74,6 +74,10 @@ public class DeviceTableScanNode extends TableScanNode {
// `pushDownLimit` row number
protected boolean pushLimitToEachDevice = false;
+ protected transient boolean containsNonAlignedDevice;
+
+ protected DeviceTableScanNode() {}
+
public DeviceTableScanNode(
PlanNodeId id,
QualifiedObjectName qualifiedObjectName,
@@ -96,7 +100,8 @@ public class DeviceTableScanNode extends TableScanNode {
Expression pushDownPredicate,
long pushDownLimit,
long pushDownOffset,
- boolean pushLimitToEachDevice) {
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice) {
super(
id,
qualifiedObjectName,
@@ -111,6 +116,7 @@ public class DeviceTableScanNode extends TableScanNode {
this.timePredicate = timePredicate;
this.pushDownPredicate = pushDownPredicate;
this.pushLimitToEachDevice = pushLimitToEachDevice;
+ this.containsNonAlignedDevice = containsNonAlignedDevice;
}
@Override
@@ -132,142 +138,75 @@ public class DeviceTableScanNode extends TableScanNode {
pushDownPredicate,
pushDownLimit,
pushDownOffset,
- pushLimitToEachDevice);
+ pushLimitToEachDevice,
+ containsNonAlignedDevice);
}
- @Override
- protected void serializeAttributes(ByteBuffer byteBuffer) {
- PlanNodeType.DEVICE_TABLE_SCAN_NODE.serialize(byteBuffer);
-
- if (qualifiedObjectName.getDatabaseName() != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), byteBuffer);
+ protected static void serializeMemberVariables(
+ DeviceTableScanNode node, ByteBuffer byteBuffer, boolean serializeOutputSymbols) {
+ TableScanNode.serializeMemberVariables(node, byteBuffer, serializeOutputSymbols);
- ReadWriteIOUtils.write(outputSymbols.size(), byteBuffer);
- outputSymbols.forEach(symbol -> ReadWriteIOUtils.write(symbol.getName(), byteBuffer));
-
- ReadWriteIOUtils.write(assignments.size(), byteBuffer);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), byteBuffer);
- ColumnSchema.serialize(entry.getValue(), byteBuffer);
- }
-
- ReadWriteIOUtils.write(deviceEntries.size(), byteBuffer);
- for (DeviceEntry entry : deviceEntries) {
+ ReadWriteIOUtils.write(node.deviceEntries.size(), byteBuffer);
+ for (DeviceEntry entry : node.deviceEntries) {
entry.serialize(byteBuffer);
}
- ReadWriteIOUtils.write(idAndAttributeIndexMap.size(), byteBuffer);
- for (Map.Entry<Symbol, Integer> entry : idAndAttributeIndexMap.entrySet()) {
+ ReadWriteIOUtils.write(node.idAndAttributeIndexMap.size(), byteBuffer);
+ for (Map.Entry<Symbol, Integer> entry : node.idAndAttributeIndexMap.entrySet()) {
Symbol.serialize(entry.getKey(), byteBuffer);
ReadWriteIOUtils.write(entry.getValue(), byteBuffer);
}
- ReadWriteIOUtils.write(scanOrder.ordinal(), byteBuffer);
-
- if (timePredicate != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- Expression.serialize(timePredicate, byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
+ ReadWriteIOUtils.write(node.scanOrder.ordinal(), byteBuffer);
- if (pushDownPredicate != null) {
+ if (node.timePredicate != null) {
ReadWriteIOUtils.write(true, byteBuffer);
- Expression.serialize(pushDownPredicate, byteBuffer);
+ Expression.serialize(node.timePredicate, byteBuffer);
} else {
ReadWriteIOUtils.write(false, byteBuffer);
}
- ReadWriteIOUtils.write(pushDownLimit, byteBuffer);
- ReadWriteIOUtils.write(pushDownOffset, byteBuffer);
- ReadWriteIOUtils.write(pushLimitToEachDevice, byteBuffer);
+ ReadWriteIOUtils.write(node.pushLimitToEachDevice, byteBuffer);
}
- @Override
- protected void serializeAttributes(DataOutputStream stream) throws IOException {
- PlanNodeType.DEVICE_TABLE_SCAN_NODE.serialize(stream);
- if (qualifiedObjectName.getDatabaseName() != null) {
- ReadWriteIOUtils.write(true, stream);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), stream);
+ protected static void serializeMemberVariables(
+ DeviceTableScanNode node, DataOutputStream stream, boolean serializeOutputSymbols)
+ throws IOException {
+ TableScanNode.serializeMemberVariables(node, stream, serializeOutputSymbols);
- ReadWriteIOUtils.write(outputSymbols.size(), stream);
- for (Symbol symbol : outputSymbols) {
- ReadWriteIOUtils.write(symbol.getName(), stream);
- }
-
- ReadWriteIOUtils.write(assignments.size(), stream);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), stream);
- ColumnSchema.serialize(entry.getValue(), stream);
- }
-
- ReadWriteIOUtils.write(deviceEntries.size(), stream);
- for (DeviceEntry entry : deviceEntries) {
+ ReadWriteIOUtils.write(node.deviceEntries.size(), stream);
+ for (DeviceEntry entry : node.deviceEntries) {
entry.serialize(stream);
}
- ReadWriteIOUtils.write(idAndAttributeIndexMap.size(), stream);
- for (Map.Entry<Symbol, Integer> entry : idAndAttributeIndexMap.entrySet()) {
+ ReadWriteIOUtils.write(node.idAndAttributeIndexMap.size(), stream);
+ for (Map.Entry<Symbol, Integer> entry : node.idAndAttributeIndexMap.entrySet()) {
Symbol.serialize(entry.getKey(), stream);
ReadWriteIOUtils.write(entry.getValue(), stream);
}
- ReadWriteIOUtils.write(scanOrder.ordinal(), stream);
+ ReadWriteIOUtils.write(node.scanOrder.ordinal(), stream);
- if (timePredicate != null) {
+ if (node.timePredicate != null) {
ReadWriteIOUtils.write(true, stream);
- Expression.serialize(timePredicate, stream);
+ Expression.serialize(node.timePredicate, stream);
} else {
ReadWriteIOUtils.write(false, stream);
}
- if (pushDownPredicate != null) {
- ReadWriteIOUtils.write(true, stream);
- Expression.serialize(pushDownPredicate, stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
-
- ReadWriteIOUtils.write(pushDownLimit, stream);
- ReadWriteIOUtils.write(pushDownOffset, stream);
- ReadWriteIOUtils.write(pushLimitToEachDevice, stream);
+ ReadWriteIOUtils.write(node.pushLimitToEachDevice, stream);
}
- public static DeviceTableScanNode deserialize(ByteBuffer byteBuffer) {
- boolean hasDatabaseName = ReadWriteIOUtils.readBool(byteBuffer);
- String databaseName = null;
- if (hasDatabaseName) {
- databaseName = ReadWriteIOUtils.readString(byteBuffer);
- }
- String tableName = ReadWriteIOUtils.readString(byteBuffer);
- QualifiedObjectName qualifiedObjectName = new QualifiedObjectName(databaseName, tableName);
+ protected static void deserializeMemberVariables(
+ ByteBuffer byteBuffer, DeviceTableScanNode node, boolean deserializeOutputSymbols) {
+ TableScanNode.deserializeMemberVariables(byteBuffer, node, deserializeOutputSymbols);
int size = ReadWriteIOUtils.readInt(byteBuffer);
- List<Symbol> outputSymbols = new ArrayList<>(size);
- for (int i = 0; i < size; i++) {
- outputSymbols.add(Symbol.deserialize(byteBuffer));
- }
-
- size = ReadWriteIOUtils.readInt(byteBuffer);
- Map<Symbol, ColumnSchema> assignments = new HashMap<>(size);
- for (int i = 0; i < size; i++) {
- assignments.put(Symbol.deserialize(byteBuffer), ColumnSchema.deserialize(byteBuffer));
- }
-
- size = ReadWriteIOUtils.readInt(byteBuffer);
List<DeviceEntry> deviceEntries = new ArrayList<>(size);
while (size-- > 0) {
deviceEntries.add(DeviceEntry.deserialize(byteBuffer));
}
+ node.deviceEntries = deviceEntries;
size = ReadWriteIOUtils.readInt(byteBuffer);
Map<Symbol, Integer> idAndAttributeIndexMap = new HashMap<>(size);
@@ -275,40 +214,38 @@ public class DeviceTableScanNode extends TableScanNode {
idAndAttributeIndexMap.put(
Symbol.deserialize(byteBuffer), ReadWriteIOUtils.readInt(byteBuffer));
}
+ node.idAndAttributeIndexMap = idAndAttributeIndexMap;
- Ordering scanOrder = Ordering.values()[ReadWriteIOUtils.readInt(byteBuffer)];
+ node.scanOrder = Ordering.values()[ReadWriteIOUtils.readInt(byteBuffer)];
- Expression timePredicate = null;
boolean hasTimePredicate = ReadWriteIOUtils.readBool(byteBuffer);
if (hasTimePredicate) {
- timePredicate = Expression.deserialize(byteBuffer);
+ node.timePredicate = Expression.deserialize(byteBuffer);
}
- Expression pushDownPredicate = null;
- boolean hasPushDownPredicate = ReadWriteIOUtils.readBool(byteBuffer);
- if (hasPushDownPredicate) {
- pushDownPredicate = Expression.deserialize(byteBuffer);
- }
+ node.pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer);
+ }
- long pushDownLimit = ReadWriteIOUtils.readLong(byteBuffer);
- long pushDownOffset = ReadWriteIOUtils.readLong(byteBuffer);
- boolean pushLimitToEachDevice = ReadWriteIOUtils.readBool(byteBuffer);
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.DEVICE_TABLE_SCAN_NODE.serialize(byteBuffer);
- PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ DeviceTableScanNode.serializeMemberVariables(this, byteBuffer, true);
+ }
- return new DeviceTableScanNode(
- planNodeId,
- qualifiedObjectName,
- outputSymbols,
- assignments,
- deviceEntries,
- idAndAttributeIndexMap,
- scanOrder,
- timePredicate,
- pushDownPredicate,
- pushDownLimit,
- pushDownOffset,
- pushLimitToEachDevice);
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.DEVICE_TABLE_SCAN_NODE.serialize(stream);
+
+ DeviceTableScanNode.serializeMemberVariables(this, stream, true);
+ }
+
+ public static DeviceTableScanNode deserialize(ByteBuffer byteBuffer) {
+ DeviceTableScanNode node = new DeviceTableScanNode();
+ DeviceTableScanNode.deserializeMemberVariables(byteBuffer, node, true);
+
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
}
public void setDeviceEntries(List<DeviceEntry> deviceEntries) {
@@ -358,4 +295,16 @@ public class DeviceTableScanNode extends TableScanNode {
public void setTimeFilter(Filter timeFilter) {
this.timeFilter = timeFilter;
}
+
+ public boolean containsNonAlignedDevice() {
+ return containsNonAlignedDevice;
+ }
+
+ public void setContainsNonAlignedDevice() {
+ this.containsNonAlignedDevice = true;
+ }
+
+ public String toString() {
+ return "DeviceTableScanNode-" + this.getPlanNodeId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/InformationSchemaTableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/InformationSchemaTableScanNode.java
index 09ff95001a6..08a11adbbdb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/InformationSchemaTableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/InformationSchemaTableScanNode.java
@@ -28,13 +28,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectN
import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
-import org.apache.tsfile.utils.ReadWriteIOUtils;
-
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -85,6 +81,8 @@ public class InformationSchemaTableScanNode extends TableScanNode {
this.regionReplicaSet = regionReplicaSet;
}
+ private InformationSchemaTableScanNode() {}
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitInformationSchemaTableScan(this, context);
@@ -107,106 +105,25 @@ public class InformationSchemaTableScanNode extends TableScanNode {
protected void serializeAttributes(ByteBuffer byteBuffer) {
PlanNodeType.INFORMATION_SCHEMA_TABLE_SCAN_NODE.serialize(byteBuffer);
- if (qualifiedObjectName.getDatabaseName() != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), byteBuffer);
-
- ReadWriteIOUtils.write(outputSymbols.size(), byteBuffer);
- outputSymbols.forEach(symbol -> ReadWriteIOUtils.write(symbol.getName(), byteBuffer));
-
- ReadWriteIOUtils.write(assignments.size(), byteBuffer);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), byteBuffer);
- ColumnSchema.serialize(entry.getValue(), byteBuffer);
- }
-
- if (pushDownPredicate != null) {
- ReadWriteIOUtils.write(true, byteBuffer);
- Expression.serialize(pushDownPredicate, byteBuffer);
- } else {
- ReadWriteIOUtils.write(false, byteBuffer);
- }
-
- ReadWriteIOUtils.write(pushDownLimit, byteBuffer);
- ReadWriteIOUtils.write(pushDownOffset, byteBuffer);
+ TableScanNode.serializeMemberVariables(this, byteBuffer, true);
}
@Override
protected void serializeAttributes(DataOutputStream stream) throws IOException {
PlanNodeType.INFORMATION_SCHEMA_TABLE_SCAN_NODE.serialize(stream);
- if (qualifiedObjectName.getDatabaseName() != null) {
- ReadWriteIOUtils.write(true, stream);
- ReadWriteIOUtils.write(qualifiedObjectName.getDatabaseName(), stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
- ReadWriteIOUtils.write(qualifiedObjectName.getObjectName(), stream);
- ReadWriteIOUtils.write(outputSymbols.size(), stream);
- for (Symbol symbol : outputSymbols) {
- ReadWriteIOUtils.write(symbol.getName(), stream);
- }
-
- ReadWriteIOUtils.write(assignments.size(), stream);
- for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
- Symbol.serialize(entry.getKey(), stream);
- ColumnSchema.serialize(entry.getValue(), stream);
- }
-
- if (pushDownPredicate != null) {
- ReadWriteIOUtils.write(true, stream);
- Expression.serialize(pushDownPredicate, stream);
- } else {
- ReadWriteIOUtils.write(false, stream);
- }
-
- ReadWriteIOUtils.write(pushDownLimit, stream);
- ReadWriteIOUtils.write(pushDownOffset, stream);
+ TableScanNode.serializeMemberVariables(this, stream, true);
}
public static InformationSchemaTableScanNode deserialize(ByteBuffer byteBuffer) {
- boolean hasDatabaseName = ReadWriteIOUtils.readBool(byteBuffer);
- String databaseName = null;
- if (hasDatabaseName) {
- databaseName = ReadWriteIOUtils.readString(byteBuffer);
- }
- String tableName = ReadWriteIOUtils.readString(byteBuffer);
- QualifiedObjectName qualifiedObjectName = new QualifiedObjectName(databaseName, tableName);
-
- int size = ReadWriteIOUtils.readInt(byteBuffer);
- List<Symbol> outputSymbols = new ArrayList<>(size);
- for (int i = 0; i < size; i++) {
- outputSymbols.add(Symbol.deserialize(byteBuffer));
- }
-
- size = ReadWriteIOUtils.readInt(byteBuffer);
- Map<Symbol, ColumnSchema> assignments = new HashMap<>(size);
- for (int i = 0; i < size; i++) {
- assignments.put(Symbol.deserialize(byteBuffer), ColumnSchema.deserialize(byteBuffer));
- }
+ InformationSchemaTableScanNode node = new InformationSchemaTableScanNode();
+ TableScanNode.deserializeMemberVariables(byteBuffer, node, true);
- Expression pushDownPredicate = null;
- boolean hasPushDownPredicate = ReadWriteIOUtils.readBool(byteBuffer);
- if (hasPushDownPredicate) {
- pushDownPredicate = Expression.deserialize(byteBuffer);
- }
-
- long pushDownLimit = ReadWriteIOUtils.readLong(byteBuffer);
- long pushDownOffset = ReadWriteIOUtils.readLong(byteBuffer);
-
- PlanNodeId planNodeId = PlanNodeId.deserialize(byteBuffer);
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
+ }
- return new InformationSchemaTableScanNode(
- planNodeId,
- qualifiedObjectName,
- outputSymbols,
- assignments,
- pushDownPredicate,
- pushDownLimit,
- pushDownOffset);
+ public String toString() {
+ return "InformationSchemaTableScanNode-" + this.getPlanNodeId();
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
index 8c6dcf33129..11807d9ceb4 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
@@ -33,9 +33,15 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import com.google.common.collect.ImmutableList;
+import org.apache.tsfile.utils.ReadWriteIOUtils;
import javax.annotation.Nullable;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -48,7 +54,7 @@ import static org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory
public abstract class TableScanNode extends SourceNode {
- protected final QualifiedObjectName qualifiedObjectName;
+ protected QualifiedObjectName qualifiedObjectName;
// Indicate the column this node need to output
protected List<Symbol> outputSymbols;
// Indicate the column this node need to fetch from StorageEngine,
@@ -97,6 +103,8 @@ public abstract class TableScanNode extends SourceNode {
this.pushDownOffset = pushDownOffset;
}
+ protected TableScanNode() {}
+
@Override
public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
return visitor.visitTableScan(this, context);
@@ -141,6 +149,10 @@ public abstract class TableScanNode extends SourceNode {
}
public Optional<Symbol> getTimeColumn() {
+ return getTimeColumn(assignments);
+ }
+
+ public static Optional<Symbol> getTimeColumn(Map<Symbol, ColumnSchema> assignments) {
for (Map.Entry<Symbol, ColumnSchema> entry : assignments.entrySet()) {
if (entry.getValue().getColumnCategory() == TIME) {
return Optional.of(entry.getKey());
@@ -213,6 +225,110 @@ public abstract class TableScanNode extends SourceNode {
this.regionReplicaSet = regionReplicaSet;
}
+ protected static void serializeMemberVariables(
+ TableScanNode node, ByteBuffer byteBuffer, boolean serializeOutputSymbols) {
+ if (node.qualifiedObjectName.getDatabaseName() != null) {
+ ReadWriteIOUtils.write(true, byteBuffer);
+ ReadWriteIOUtils.write(node.qualifiedObjectName.getDatabaseName(), byteBuffer);
+ } else {
+ ReadWriteIOUtils.write(false, byteBuffer);
+ }
+ ReadWriteIOUtils.write(node.qualifiedObjectName.getObjectName(), byteBuffer);
+
+ if (serializeOutputSymbols) {
+ ReadWriteIOUtils.write(node.outputSymbols.size(), byteBuffer);
+ node.outputSymbols.forEach(symbol -> ReadWriteIOUtils.write(symbol.getName(), byteBuffer));
+ }
+
+ ReadWriteIOUtils.write(node.assignments.size(), byteBuffer);
+ for (Map.Entry<Symbol, ColumnSchema> entry : node.assignments.entrySet()) {
+ Symbol.serialize(entry.getKey(), byteBuffer);
+ ColumnSchema.serialize(entry.getValue(), byteBuffer);
+ }
+
+ if (node.pushDownPredicate != null) {
+ ReadWriteIOUtils.write(true, byteBuffer);
+ Expression.serialize(node.pushDownPredicate, byteBuffer);
+ } else {
+ ReadWriteIOUtils.write(false, byteBuffer);
+ }
+
+ ReadWriteIOUtils.write(node.pushDownLimit, byteBuffer);
+ ReadWriteIOUtils.write(node.pushDownOffset, byteBuffer);
+ }
+
+ protected static void serializeMemberVariables(
+ TableScanNode node, DataOutputStream stream, boolean serializeOutputSymbols)
+ throws IOException {
+ if (node.qualifiedObjectName.getDatabaseName() != null) {
+ ReadWriteIOUtils.write(true, stream);
+ ReadWriteIOUtils.write(node.qualifiedObjectName.getDatabaseName(), stream);
+ } else {
+ ReadWriteIOUtils.write(false, stream);
+ }
+ ReadWriteIOUtils.write(node.qualifiedObjectName.getObjectName(), stream);
+
+ if (serializeOutputSymbols) {
+ ReadWriteIOUtils.write(node.outputSymbols.size(), stream);
+ for (Symbol symbol : node.outputSymbols) {
+ ReadWriteIOUtils.write(symbol.getName(), stream);
+ }
+ }
+
+ ReadWriteIOUtils.write(node.assignments.size(), stream);
+ for (Map.Entry<Symbol, ColumnSchema> entry : node.assignments.entrySet()) {
+ Symbol.serialize(entry.getKey(), stream);
+ ColumnSchema.serialize(entry.getValue(), stream);
+ }
+
+ if (node.pushDownPredicate != null) {
+ ReadWriteIOUtils.write(true, stream);
+ Expression.serialize(node.pushDownPredicate, stream);
+ } else {
+ ReadWriteIOUtils.write(false, stream);
+ }
+
+ ReadWriteIOUtils.write(node.pushDownLimit, stream);
+ ReadWriteIOUtils.write(node.pushDownOffset, stream);
+ }
+
+ protected static void deserializeMemberVariables(
+ ByteBuffer byteBuffer, TableScanNode node, boolean deserializeOutputSymbols) {
+ boolean hasDatabaseName = ReadWriteIOUtils.readBool(byteBuffer);
+ String databaseName = null;
+ if (hasDatabaseName) {
+ databaseName = ReadWriteIOUtils.readString(byteBuffer);
+ }
+ String tableName = ReadWriteIOUtils.readString(byteBuffer);
+ node.qualifiedObjectName = new QualifiedObjectName(databaseName, tableName);
+
+ int size;
+
+ if (deserializeOutputSymbols) {
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ List<Symbol> outputSymbols = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ outputSymbols.add(Symbol.deserialize(byteBuffer));
+ }
+ node.outputSymbols = outputSymbols;
+ }
+
+ size = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<Symbol, ColumnSchema> assignments = new HashMap<>(size);
+ for (int i = 0; i < size; i++) {
+ assignments.put(Symbol.deserialize(byteBuffer), ColumnSchema.deserialize(byteBuffer));
+ }
+ node.assignments = assignments;
+
+ boolean hasPushDownPredicate = ReadWriteIOUtils.readBool(byteBuffer);
+ if (hasPushDownPredicate) {
+ node.pushDownPredicate = Expression.deserialize(byteBuffer);
+ }
+
+ node.pushDownLimit = ReadWriteIOUtils.readLong(byteBuffer);
+ node.pushDownOffset = ReadWriteIOUtils.readLong(byteBuffer);
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java
new file mode 100644
index 00000000000..f6ee41c54bf
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeAlignedDeviceViewScanNode.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.planner.node;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+
+public class TreeAlignedDeviceViewScanNode extends TreeDeviceViewScanNode {
+
+ public TreeAlignedDeviceViewScanNode(
+ PlanNodeId id,
+ QualifiedObjectName qualifiedObjectName,
+ List<Symbol> outputSymbols,
+ Map<Symbol, ColumnSchema> assignments,
+ List<DeviceEntry> deviceEntries,
+ Map<Symbol, Integer> idAndAttributeIndexMap,
+ Ordering scanOrder,
+ Expression timePredicate,
+ Expression pushDownPredicate,
+ long pushDownLimit,
+ long pushDownOffset,
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice,
+ String treeDBName,
+ Map<String, String> measurementColumnNameMap) {
+ super(
+ id,
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ public TreeAlignedDeviceViewScanNode() {}
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitTreeAlignedDeviceViewScan(this, context);
+ }
+
+ @Override
+ public TreeAlignedDeviceViewScanNode clone() {
+ return new TreeAlignedDeviceViewScanNode(
+ getPlanNodeId(),
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TREE_ALIGNED_DEVICE_VIEW_SCAN_NODE.serialize(byteBuffer);
+
+ TreeDeviceViewScanNode.serializeMemberVariables(this, byteBuffer);
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TREE_ALIGNED_DEVICE_VIEW_SCAN_NODE.serialize(stream);
+
+ TreeDeviceViewScanNode.serializeMemberVariables(this, stream);
+ }
+
+ public static TreeAlignedDeviceViewScanNode deserialize(ByteBuffer byteBuffer) {
+ TreeAlignedDeviceViewScanNode node = new TreeAlignedDeviceViewScanNode();
+ TreeDeviceViewScanNode.deserializeMemberVariables(byteBuffer, node, true);
+
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
+ }
+
+ public String toString() {
+ return "TreeAlignedDeviceViewScanNode-" + this.getPlanNodeId();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java
new file mode 100644
index 00000000000..8e0a08f2830
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeDeviceViewScanNode.java
@@ -0,0 +1,208 @@
+/*
+ * 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.plan.relational.planner.node;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+public class TreeDeviceViewScanNode extends DeviceTableScanNode {
+ protected String treeDBName;
+ protected Map<String, String> measurementColumnNameMap;
+
+ public TreeDeviceViewScanNode(
+ PlanNodeId id,
+ QualifiedObjectName qualifiedObjectName,
+ List<Symbol> outputSymbols,
+ Map<Symbol, ColumnSchema> assignments,
+ Map<Symbol, Integer> idAndAttributeIndexMap,
+ String treeDBName,
+ Map<String, String> measurementColumnNameMap) {
+ super(id, qualifiedObjectName, outputSymbols, assignments, idAndAttributeIndexMap);
+ this.treeDBName = treeDBName;
+ this.measurementColumnNameMap = measurementColumnNameMap;
+ }
+
+ public TreeDeviceViewScanNode(
+ PlanNodeId id,
+ QualifiedObjectName qualifiedObjectName,
+ List<Symbol> outputSymbols,
+ Map<Symbol, ColumnSchema> assignments,
+ List<DeviceEntry> deviceEntries,
+ Map<Symbol, Integer> idAndAttributeIndexMap,
+ Ordering scanOrder,
+ Expression timePredicate,
+ Expression pushDownPredicate,
+ long pushDownLimit,
+ long pushDownOffset,
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice,
+ String treeDBName,
+ Map<String, String> measurementColumnNameMap) {
+ super(
+ id,
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice);
+ this.treeDBName = treeDBName;
+ this.measurementColumnNameMap = measurementColumnNameMap;
+ }
+
+ public TreeDeviceViewScanNode() {}
+
+ public String getTreeDBName() {
+ return treeDBName;
+ }
+
+ public Map<String, String> getMeasurementColumnNameMap() {
+ return measurementColumnNameMap;
+ }
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitTreeDeviceViewScan(this, context);
+ }
+
+ @Override
+ public TreeDeviceViewScanNode clone() {
+ return new TreeDeviceViewScanNode(
+ getPlanNodeId(),
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ protected static void serializeMemberVariables(
+ TreeDeviceViewScanNode node, ByteBuffer byteBuffer) {
+ DeviceTableScanNode.serializeMemberVariables(node, byteBuffer, true);
+
+ ReadWriteIOUtils.write(node.treeDBName, byteBuffer);
+ ReadWriteIOUtils.write(node.measurementColumnNameMap.size(), byteBuffer);
+ for (Map.Entry<String, String> entry : node.measurementColumnNameMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), byteBuffer);
+ ReadWriteIOUtils.write(entry.getValue(), byteBuffer);
+ }
+ }
+
+ protected static void serializeMemberVariables(
+ TreeDeviceViewScanNode node, DataOutputStream stream) throws IOException {
+ DeviceTableScanNode.serializeMemberVariables(node, stream, true);
+
+ ReadWriteIOUtils.write(node.treeDBName, stream);
+ ReadWriteIOUtils.write(node.measurementColumnNameMap.size(), stream);
+ for (Map.Entry<String, String> entry : node.measurementColumnNameMap.entrySet()) {
+ ReadWriteIOUtils.write(entry.getKey(), stream);
+ ReadWriteIOUtils.write(entry.getValue(), stream);
+ }
+ }
+
+ protected static void deserializeMemberVariables(
+ ByteBuffer byteBuffer, TreeDeviceViewScanNode node) {
+ DeviceTableScanNode.deserializeMemberVariables(byteBuffer, node, true);
+
+ node.treeDBName = ReadWriteIOUtils.readString(byteBuffer);
+ int size = ReadWriteIOUtils.readInt(byteBuffer);
+ Map<String, String> measurementColumnNameMap = new HashMap<>(size);
+ for (int i = 0; i < size; i++) {
+ measurementColumnNameMap.put(
+ ReadWriteIOUtils.readString(byteBuffer), ReadWriteIOUtils.readString(byteBuffer));
+ }
+ node.measurementColumnNameMap = measurementColumnNameMap;
+ }
+
+ // We will transform this node into its sub-Class when generate DistributionPlan, so it will never
+ // should be serialized or deserialized.
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ throw new UnsupportedOperationException(
+ "Unsupported to serialize: " + TreeNonAlignedDeviceViewScanNode.class.getSimpleName());
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ throw new UnsupportedOperationException(
+ "Unsupported to serialize: " + TreeNonAlignedDeviceViewScanNode.class.getSimpleName());
+ }
+
+ public static DeviceTableScanNode deserialize(ByteBuffer byteBuffer) {
+ throw new UnsupportedOperationException(
+ "Unsupported to deserialize: " + TreeNonAlignedDeviceViewScanNode.class.getSimpleName());
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ TreeDeviceViewScanNode that = (TreeDeviceViewScanNode) o;
+ return Objects.equals(treeDBName, that.treeDBName)
+ && Objects.equals(measurementColumnNameMap, that.measurementColumnNameMap);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), treeDBName, measurementColumnNameMap);
+ }
+
+ public String toString() {
+ return "TreeDeviceViewScanNode-" + this.getPlanNodeId();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java
new file mode 100644
index 00000000000..f2202feda47
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.queryengine.plan.relational.planner.node;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeType;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanVisitor;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.Map;
+
+public class TreeNonAlignedDeviceViewScanNode extends TreeDeviceViewScanNode {
+
+ public TreeNonAlignedDeviceViewScanNode(
+ PlanNodeId id,
+ QualifiedObjectName qualifiedObjectName,
+ List<Symbol> outputSymbols,
+ Map<Symbol, ColumnSchema> assignments,
+ List<DeviceEntry> deviceEntries,
+ Map<Symbol, Integer> idAndAttributeIndexMap,
+ Ordering scanOrder,
+ Expression timePredicate,
+ Expression pushDownPredicate,
+ long pushDownLimit,
+ long pushDownOffset,
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice,
+ String treeDBName,
+ Map<String, String> measurementColumnNameMap) {
+ super(
+ id,
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ public TreeNonAlignedDeviceViewScanNode() {}
+
+ @Override
+ public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
+ return visitor.visitTreeNonAlignedDeviceViewScan(this, context);
+ }
+
+ @Override
+ public TreeNonAlignedDeviceViewScanNode clone() {
+ return new TreeNonAlignedDeviceViewScanNode(
+ getPlanNodeId(),
+ qualifiedObjectName,
+ outputSymbols,
+ assignments,
+ deviceEntries,
+ idAndAttributeIndexMap,
+ scanOrder,
+ timePredicate,
+ pushDownPredicate,
+ pushDownLimit,
+ pushDownOffset,
+ pushLimitToEachDevice,
+ containsNonAlignedDevice,
+ treeDBName,
+ measurementColumnNameMap);
+ }
+
+ @Override
+ protected void serializeAttributes(ByteBuffer byteBuffer) {
+ PlanNodeType.TREE_NONALIGNED_DEVICE_VIEW_SCAN_NODE.serialize(byteBuffer);
+
+ TreeDeviceViewScanNode.serializeMemberVariables(this, byteBuffer);
+ }
+
+ @Override
+ protected void serializeAttributes(DataOutputStream stream) throws IOException {
+ PlanNodeType.TREE_NONALIGNED_DEVICE_VIEW_SCAN_NODE.serialize(stream);
+
+ TreeDeviceViewScanNode.serializeMemberVariables(this, stream);
+ }
+
+ public static TreeNonAlignedDeviceViewScanNode deserialize(ByteBuffer byteBuffer) {
+ TreeNonAlignedDeviceViewScanNode node = new TreeNonAlignedDeviceViewScanNode();
+ TreeDeviceViewScanNode.deserializeMemberVariables(byteBuffer, node, true);
+
+ node.setPlanNodeId(PlanNodeId.deserialize(byteBuffer));
+ return node;
+ }
+
+ public String toString() {
+ return "TreeNonAlignedDeviceViewScanNode-" + this.getPlanNodeId();
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
index c1473c1a333..9d753386394 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushAggregationIntoTableScan.java
@@ -105,6 +105,10 @@ public class PushAggregationIntoTableScan implements PlanOptimizer {
return node;
}
+ if (tableScanNode.containsNonAlignedDevice()) {
+ return node;
+ }
+
PushDownLevel pushDownLevel =
calculatePushDownLevel(
node.getAggregations().values(),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
index fefdbf6f605..79da48a8cce 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/PushPredicateIntoTableScan.java
@@ -38,6 +38,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.analyzer.predicate.Predic
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.NonAlignedAlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.planner.Assignments;
import org.apache.iotdb.db.queryengine.plan.relational.planner.EqualityInference;
import org.apache.iotdb.db.queryengine.plan.relational.planner.OrderingScheme;
@@ -51,6 +52,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.FilterNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.JoinNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ComparisonExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.FunctionCall;
@@ -449,6 +451,10 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
attributeColumns,
queryContext);
tableScanNode.setDeviceEntries(deviceEntries);
+ if (deviceEntries.stream()
+ .anyMatch(deviceEntry -> deviceEntry instanceof NonAlignedAlignedDeviceEntry)) {
+ tableScanNode.setContainsNonAlignedDevice();
+ }
final long schemaFetchCost = System.nanoTime() - startTime;
QueryPlanCostMetricSet.getInstance()
@@ -473,7 +479,10 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
startTime = System.nanoTime();
final DataPartition dataPartition =
fetchDataPartitionByDevices(
- tableScanNode.getQualifiedObjectName().getDatabaseName(),
+ // for tree view, we need to pass actual tree db name to this method
+ tableScanNode instanceof TreeDeviceViewScanNode
+ ? ((TreeDeviceViewScanNode) tableScanNode).getTreeDBName()
+ : tableScanNode.getQualifiedObjectName().getDatabaseName(),
deviceEntries,
timeFilter);
@@ -754,7 +763,8 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
}
private DataPartition fetchDataPartitionByDevices(
- final String database,
+ final String
+ database, // for tree view, database should be the real tree db name with `root.` prefix
final List<DeviceEntry> deviceEntries,
final Filter globalTimeFilter) {
final Pair<List<TTimePartitionSlot>, Pair<Boolean, Boolean>> res =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
index 348eb7823ee..dda33ea1c2f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/UnaliasSymbolReferences.java
@@ -47,6 +47,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.PreviousFill
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TopKNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ValueFillNode;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.NullLiteral;
@@ -151,6 +152,42 @@ public class UnaliasSymbolReferences implements PlanOptimizer {
return new PlanAndMappings(rewrittenAggregation, mapping);
}
+ @Override
+ public PlanAndMappings visitTreeDeviceViewScan(
+ TreeDeviceViewScanNode node, UnaliasContext context) {
+ Map<Symbol, Symbol> mapping = new HashMap<>(context.getCorrelationMapping());
+ SymbolMapper mapper = symbolMapper(mapping);
+
+ List<Symbol> newOutputs = mapper.map(node.getOutputSymbols());
+
+ Map<Symbol, ColumnSchema> newAssignments = new HashMap<>();
+ node.getAssignments()
+ .forEach(
+ (symbol, handle) -> {
+ Symbol newSymbol = mapper.map(symbol);
+ newAssignments.put(newSymbol, handle);
+ });
+
+ return new PlanAndMappings(
+ new TreeDeviceViewScanNode(
+ node.getPlanNodeId(),
+ node.getQualifiedObjectName(),
+ newOutputs,
+ newAssignments,
+ node.getDeviceEntries(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice(),
+ node.getTreeDBName(),
+ node.getMeasurementColumnNameMap()),
+ mapping);
+ }
+
@Override
public PlanAndMappings visitDeviceTableScan(DeviceTableScanNode node, UnaliasContext context) {
Map<Symbol, Symbol> mapping = new HashMap<>(context.getCorrelationMapping());
@@ -179,7 +216,8 @@ public class UnaliasSymbolReferences implements PlanOptimizer {
node.getPushDownPredicate(),
node.getPushDownLimit(),
node.getPushDownOffset(),
- node.isPushLimitToEachDevice()),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice()),
mapping);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java
index 8fff6f2eb23..83798c451bc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/optimizations/Util.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.SymbolAllocator;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import com.google.common.collect.ImmutableList;
import org.apache.tsfile.read.common.type.Type;
@@ -115,6 +116,13 @@ public class Util {
*/
public static Pair<AggregationNode, AggregationTableScanNode> split(
AggregationTableScanNode node, SymbolAllocator symbolAllocator, QueryId queryId) {
+ AggregationTreeDeviceViewScanNode aggregationTreeDeviceViewScanNode;
+ if (node instanceof AggregationTreeDeviceViewScanNode) {
+ aggregationTreeDeviceViewScanNode = (AggregationTreeDeviceViewScanNode) node;
+ } else {
+ aggregationTreeDeviceViewScanNode = null;
+ }
+
Map<Symbol, AggregationNode.Aggregation> intermediateAggregation = new LinkedHashMap<>();
Map<Symbol, AggregationNode.Aggregation> finalAggregation = new LinkedHashMap<>();
for (Map.Entry<Symbol, AggregationNode.Aggregation> entry : node.getAggregations().entrySet()) {
@@ -152,6 +160,51 @@ public class Util {
Optional.empty()));
}
+ AggregationTableScanNode rightResult =
+ aggregationTreeDeviceViewScanNode == null
+ ? new AggregationTableScanNode(
+ queryId.genPlanNodeId(),
+ node.getQualifiedObjectName(),
+ node.getOutputSymbols(),
+ node.getAssignments(),
+ ImmutableList.of(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice(),
+ node.getProjection(),
+ intermediateAggregation,
+ node.getGroupingSets(),
+ node.getPreGroupedSymbols(),
+ PARTIAL,
+ node.getGroupIdSymbol())
+ : new AggregationTreeDeviceViewScanNode(
+ queryId.genPlanNodeId(),
+ node.getQualifiedObjectName(),
+ node.getOutputSymbols(),
+ node.getAssignments(),
+ ImmutableList.of(),
+ node.getIdAndAttributeIndexMap(),
+ node.getScanOrder(),
+ node.getTimePredicate().orElse(null),
+ node.getPushDownPredicate(),
+ node.getPushDownLimit(),
+ node.getPushDownOffset(),
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice(),
+ node.getProjection(),
+ intermediateAggregation,
+ node.getGroupingSets(),
+ node.getPreGroupedSymbols(),
+ PARTIAL,
+ node.getGroupIdSymbol(),
+ aggregationTreeDeviceViewScanNode.getTreeDBName(),
+ aggregationTreeDeviceViewScanNode.getMeasurementColumnNameMap());
+
return new Pair<>(
new AggregationNode(
node.getPlanNodeId(),
@@ -162,24 +215,6 @@ public class Util {
FINAL,
Optional.empty(),
node.getGroupIdSymbol()),
- new AggregationTableScanNode(
- queryId.genPlanNodeId(),
- node.getQualifiedObjectName(),
- node.getOutputSymbols(),
- node.getAssignments(),
- ImmutableList.of(),
- node.getIdAndAttributeIndexMap(),
- node.getScanOrder(),
- node.getTimePredicate().orElse(null),
- node.getPushDownPredicate(),
- node.getPushDownLimit(),
- node.getPushDownOffset(),
- node.isPushLimitToEachDevice(),
- node.getProjection(),
- intermediateAggregation,
- node.getGroupingSets(),
- node.getPreGroupedSymbols(),
- PARTIAL,
- node.getGroupIdSymbol()));
+ rightResult);
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java
index c536d21f17d..24e07245e61 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/MockTableModelDataPartition.java
@@ -48,8 +48,6 @@ public class MockTableModelDataPartition {
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
- private static final String DB_NAME = "testdb";
-
static final String DEVICE_1 = "table1.beijing.A1.ZZ";
static final String DEVICE_2 = "table1.beijing.A2.XX";
static final String DEVICE_3 = "table1.shanghai.A3.YY";
@@ -99,7 +97,7 @@ public class MockTableModelDataPartition {
* device6(startTime:0): DataRegionGroup_2,
* device6(startTime:100): DataRegionGroup_3,
*/
- public static DataPartition constructDataPartition() {
+ public static DataPartition constructDataPartition(String dbName) {
DataPartition dataPartition =
new DataPartition(
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
@@ -132,13 +130,13 @@ public class MockTableModelDataPartition {
devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_5), dataRegionMap3);
devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_6), dataRegionMap3);
- dbPartitionMap.put(DB_NAME, devicePartitionMap);
+ dbPartitionMap.put(dbName, devicePartitionMap);
dataPartition.setDataPartitionMap(dbPartitionMap);
return dataPartition;
}
- public static SchemaPartition constructSchemaPartition() {
+ public static SchemaPartition constructSchemaPartition(String dbName) {
final SchemaPartition schemaPartition =
new SchemaPartition(
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
@@ -162,7 +160,7 @@ public class MockTableModelDataPartition {
schemaRegionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_1), schemaRegion1);
schemaRegionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_2), schemaRegion2);
schemaRegionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_3), schemaRegion2);
- schemaPartitionMap.put(DB_NAME, schemaRegionMap);
+ schemaPartitionMap.put(dbName, schemaRegionMap);
schemaPartition.setSchemaPartitionMap(schemaPartitionMap);
return schemaPartition;
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
index 0a665fbed04..9ff5ebfdb8c 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TSBSMetadata.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnMetadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
@@ -289,40 +290,42 @@ public class TSBSMetadata implements Metadata {
&& attributeColumns.isEmpty()) {
// r01, r02
return ImmutableList.of(
- new DeviceEntry(new StringArrayDeviceID(T1_DEVICE_1.split("\\.")), ImmutableList.of()),
- new DeviceEntry(new StringArrayDeviceID(T1_DEVICE_2.split("\\.")), ImmutableList.of()));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(T1_DEVICE_1.split("\\.")), ImmutableList.of()),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(T1_DEVICE_2.split("\\.")), ImmutableList.of()));
} else if (expressionList.size() == 1
&& expressionList.get(0).toString().equals("(\"fleet\" = 'South')")
&& attributeColumns.size() == 1
&& attributeColumns.get(0).equals("load_capacity")) {
// r03
return ImmutableList.of(
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T1_DEVICE_1.split("\\.")),
ImmutableList.of(new Binary("2000", TSFileConfig.STRING_CHARSET))),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T1_DEVICE_2.split("\\.")),
ImmutableList.of(new Binary("1000", TSFileConfig.STRING_CHARSET))));
} else {
// others (The return result maybe not correct in actual, but it is convenient for test of
// DistributionPlan)
return Arrays.asList(
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T1_DEVICE_1.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T1_DEVICE_2.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T1_DEVICE_3.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T2_DEVICE_1.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T2_DEVICE_2.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)),
- new DeviceEntry(
+ new AlignedDeviceEntry(
new StringArrayDeviceID(T2_DEVICE_3.split("\\.")),
ImmutableList.of(Binary.EMPTY_VALUE, Binary.EMPTY_VALUE)));
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
index 8d2d32cdfcd..a7ebed0ec6f 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestMatadata.java
@@ -31,14 +31,17 @@ import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.SessionInfo;
import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
import org.apache.iotdb.db.queryengine.plan.relational.function.OperatorType;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.AlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnMetadata;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ColumnSchema;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.DeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.ITableDeviceSchemaValidation;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.Metadata;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.NonAlignedAlignedDeviceEntry;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.OperatorNotFoundException;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.QualifiedObjectName;
import org.apache.iotdb.db.queryengine.plan.relational.metadata.TableSchema;
+import org.apache.iotdb.db.queryengine.plan.relational.metadata.TreeDeviceViewSchema;
import org.apache.iotdb.db.queryengine.plan.relational.security.AccessControl;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ComparisonExpression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
@@ -51,11 +54,14 @@ import org.apache.iotdb.db.queryengine.plan.relational.type.TypeSignature;
import org.apache.iotdb.db.schemaengine.table.InformationSchemaUtils;
import org.apache.iotdb.mpp.rpc.thrift.TRegionRouteReq;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.file.metadata.StringArrayDeviceID;
import org.apache.tsfile.read.common.type.StringType;
import org.apache.tsfile.read.common.type.Type;
import org.apache.tsfile.read.common.type.TypeFactory;
+import org.mockito.Mockito;
import java.util.Arrays;
import java.util.Collections;
@@ -92,6 +98,7 @@ public class TestMatadata implements Metadata {
private final TypeManager typeManager = new InternalTypeManager();
public static final String DB1 = "testdb";
+ public static final String TREE_DB1 = "root.test";
public static final String TABLE1 = "table1";
public static final String TIME = "time";
private static final String TAG1 = "tag1";
@@ -115,6 +122,9 @@ public class TestMatadata implements Metadata {
public static final String DB2 = "db2";
public static final String TABLE2 = "table2";
+ public static final String TREE_VIEW_DB = "tree_view";
+ public static final String DEVICE_VIEW_TEST_TABLE = "root.test.device_view";
+
@Override
public boolean tableExists(final QualifiedObjectName name) {
return name.getDatabaseName().equalsIgnoreCase(DB1)
@@ -123,6 +133,33 @@ public class TestMatadata implements Metadata {
@Override
public Optional<TableSchema> getTableSchema(SessionInfo session, QualifiedObjectName name) {
+ if (name.getDatabaseName().equals(TREE_VIEW_DB)) {
+ TreeDeviceViewSchema treeDeviceViewSchema = Mockito.mock(TreeDeviceViewSchema.class);
+ Mockito.when(treeDeviceViewSchema.getTableName()).thenReturn(DEVICE_VIEW_TEST_TABLE);
+ Mockito.when(treeDeviceViewSchema.getColumns())
+ .thenReturn(
+ ImmutableList.of(
+ ColumnSchema.builder(TIME_CM)
+ .setColumnCategory(TsTableColumnCategory.TIME)
+ .build(),
+ ColumnSchema.builder(TAG1_CM)
+ .setColumnCategory(TsTableColumnCategory.TAG)
+ .build(),
+ ColumnSchema.builder(TAG2_CM)
+ .setColumnCategory(TsTableColumnCategory.TAG)
+ .build(),
+ ColumnSchema.builder(S1_CM)
+ .setColumnCategory(TsTableColumnCategory.FIELD)
+ .build(),
+ ColumnSchema.builder(S2_CM)
+ .setColumnCategory(TsTableColumnCategory.FIELD)
+ .build()));
+ Mockito.when(treeDeviceViewSchema.getTreeDBName()).thenReturn(TREE_DB1);
+ Mockito.when(treeDeviceViewSchema.getMeasurementColumnNameMap())
+ .thenReturn(ImmutableMap.of(TAG1, "province", TAG2, "city"));
+ return Optional.of(treeDeviceViewSchema);
+ }
+
if (name.getDatabaseName().equals(INFORMATION_DATABASE)) {
TsTable table =
InformationSchemaUtils.mayGetTable(
@@ -238,6 +275,23 @@ public class TestMatadata implements Metadata {
final List<Expression> expressionList,
final List<String> attributeColumns,
final MPPQueryContext context) {
+ if (tableName.getDatabaseName().equals(TREE_VIEW_DB)) {
+ if (expressionList.isEmpty()) {
+ return ImmutableList.of(
+ new AlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE_3), ImmutableList.of()),
+ new AlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE_6), ImmutableList.of()),
+ new NonAlignedAlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE_5), ImmutableList.of()));
+ }
+
+ return ImmutableList.of(
+ new AlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE_3), ImmutableList.of()),
+ new AlignedDeviceEntry(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(DEVICE_6), ImmutableList.of()));
+ }
if (expressionList.size() == 2) {
if (compareEqualsMatch(expressionList.get(0), "tag1", "beijing")
@@ -245,47 +299,59 @@ public class TestMatadata implements Metadata {
|| compareEqualsMatch(expressionList.get(1), "tag1", "beijing")
&& compareEqualsMatch(expressionList.get(0), "tag2", "A1")) {
return Collections.singletonList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES));
}
if (compareEqualsMatch(expressionList.get(0), "tag1", "shanghai")
&& compareEqualsMatch(expressionList.get(1), "tag2", "B3")
|| compareEqualsMatch(expressionList.get(1), "tag1", "shanghai")
&& compareEqualsMatch(expressionList.get(0), "tag2", "B3")) {
return Collections.singletonList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_1_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_1_ATTRIBUTES));
}
} else if (expressionList.size() == 1) {
if (compareEqualsMatch(expressionList.get(0), "tag1", "shanghai")) {
return Arrays.asList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES));
}
if (compareEqualsMatch(expressionList.get(0), "tag1", "shenzhen")) {
return Arrays.asList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_6.split("\\.")), DEVICE_6_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_6.split("\\.")), DEVICE_6_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES));
}
if (compareNotEqualsMatch(expressionList.get(0), "tag1", "shenzhen")) {
return Arrays.asList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_2.split("\\.")), DEVICE_2_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_2.split("\\.")), DEVICE_2_ATTRIBUTES));
}
if (compareEqualsMatch(expressionList.get(0), "tag2", "B2")) {
return Collections.singletonList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES));
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES));
}
}
return Arrays.asList(
- new DeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_6.split("\\.")), DEVICE_6_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES),
- new DeviceEntry(new StringArrayDeviceID(DEVICE_2.split("\\.")), DEVICE_2_ATTRIBUTES));
+ new AlignedDeviceEntry(new StringArrayDeviceID(DEVICE_4.split("\\.")), DEVICE_4_ATTRIBUTES),
+ new AlignedDeviceEntry(new StringArrayDeviceID(DEVICE_1.split("\\.")), DEVICE_1_ATTRIBUTES),
+ new AlignedDeviceEntry(new StringArrayDeviceID(DEVICE_6.split("\\.")), DEVICE_6_ATTRIBUTES),
+ new AlignedDeviceEntry(new StringArrayDeviceID(DEVICE_5.split("\\.")), DEVICE_5_ATTRIBUTES),
+ new AlignedDeviceEntry(new StringArrayDeviceID(DEVICE_3.split("\\.")), DEVICE_3_ATTRIBUTES),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_2.split("\\.")), DEVICE_2_ATTRIBUTES));
}
private boolean compareEqualsMatch(
@@ -364,20 +430,26 @@ public class TestMatadata implements Metadata {
@Override
public DataPartition getDataPartition(
final String database, final List<DataPartitionQueryParam> sgNameToQueryParamsMap) {
- return DATA_PARTITION;
+ return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION;
}
@Override
public DataPartition getDataPartitionWithUnclosedTimeRange(
final String database, final List<DataPartitionQueryParam> sgNameToQueryParamsMap) {
- return DATA_PARTITION;
+ return TREE_DB1.equals(database) ? TREE_VIEW_DATA_PARTITION : TABLE_DATA_PARTITION;
}
- private static final DataPartition DATA_PARTITION =
- MockTableModelDataPartition.constructDataPartition();
+ private static final DataPartition TABLE_DATA_PARTITION =
+ MockTableModelDataPartition.constructDataPartition(DB1);
+
+ private static final DataPartition TREE_VIEW_DATA_PARTITION =
+ MockTableModelDataPartition.constructDataPartition(TREE_DB1);
+
+ private static final SchemaPartition TABLE_SCHEMA_PARTITION =
+ MockTableModelDataPartition.constructSchemaPartition(DB1);
- private static final SchemaPartition SCHEMA_PARTITION =
- MockTableModelDataPartition.constructSchemaPartition();
+ private static final SchemaPartition TREE_SCHEMA_PARTITION =
+ MockTableModelDataPartition.constructSchemaPartition(TREE_DB1);
private static IPartitionFetcher getFakePartitionFetcher() {
@@ -385,37 +457,41 @@ public class TestMatadata implements Metadata {
@Override
public SchemaPartition getSchemaPartition(PathPatternTree patternTree) {
- return SCHEMA_PARTITION;
+ return TABLE_SCHEMA_PARTITION;
}
@Override
public SchemaPartition getOrCreateSchemaPartition(
PathPatternTree patternTree, String userName) {
- return SCHEMA_PARTITION;
+ return TABLE_SCHEMA_PARTITION;
}
@Override
public DataPartition getDataPartition(
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
- return DATA_PARTITION;
+ return !sgNameToQueryParamsMap.isEmpty() && sgNameToQueryParamsMap.get(TREE_VIEW_DB) != null
+ ? TREE_VIEW_DATA_PARTITION
+ : TABLE_DATA_PARTITION;
}
@Override
public DataPartition getDataPartitionWithUnclosedTimeRange(
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
- return DATA_PARTITION;
+ return !sgNameToQueryParamsMap.isEmpty() && sgNameToQueryParamsMap.get(TREE_VIEW_DB) != null
+ ? TREE_VIEW_DATA_PARTITION
+ : TABLE_DATA_PARTITION;
}
@Override
public DataPartition getOrCreateDataPartition(
Map<String, List<DataPartitionQueryParam>> sgNameToQueryParamsMap) {
- return DATA_PARTITION;
+ return TABLE_DATA_PARTITION;
}
@Override
public DataPartition getOrCreateDataPartition(
List<DataPartitionQueryParam> dataPartitionQueryParams, String userName) {
- return DATA_PARTITION;
+ return TABLE_DATA_PARTITION;
}
@Override
@@ -435,12 +511,12 @@ public class TestMatadata implements Metadata {
@Override
public SchemaPartition getOrCreateSchemaPartition(
String database, List<IDeviceID> deviceIDList, String userName) {
- return SCHEMA_PARTITION;
+ return TREE_VIEW_DB.equals(database) ? TREE_SCHEMA_PARTITION : TABLE_SCHEMA_PARTITION;
}
@Override
public SchemaPartition getSchemaPartition(String database, List<IDeviceID> deviceIDList) {
- return SCHEMA_PARTITION;
+ return TREE_VIEW_DB.equals(database) ? TREE_SCHEMA_PARTITION : TABLE_SCHEMA_PARTITION;
}
};
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestPlanBuilder.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestPlanBuilder.java
index 1b8764f9085..ce5d7081780 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestPlanBuilder.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TestPlanBuilder.java
@@ -85,7 +85,8 @@ public class TestPlanBuilder {
Expression pushDownPredicate,
long pushDownLimit,
long pushDownOffset,
- boolean pushLimitToEachDevice) {
+ boolean pushLimitToEachDevice,
+ boolean containsNonAlignedDevice) {
this.root =
new DeviceTableScanNode(
new PlanNodeId(id),
@@ -99,7 +100,8 @@ public class TestPlanBuilder {
pushDownPredicate,
pushDownLimit,
pushDownOffset,
- pushLimitToEachDevice);
+ pushLimitToEachDevice,
+ containsNonAlignedDevice);
return this;
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
new file mode 100644
index 00000000000..f54151eaa01
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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.plan.relational.analyzer;
+
+import org.apache.iotdb.db.queryengine.plan.planner.plan.LogicalQueryPlan;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.PlanTester;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.junit.Test;
+
+import java.util.Optional;
+
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMatadata.DEVICE_VIEW_TEST_TABLE;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMatadata.TREE_VIEW_DB;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanAssert.assertPlan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.aggregation;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.aggregationFunction;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.aggregationTreeDeviceViewTableScan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.exchange;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.mergeSort;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.output;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.project;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.singleGroupingSet;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.streamSort;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.treeAlignedDeviceViewTableScan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.treeDeviceViewTableScan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.assertions.PlanMatchPattern.treeNonAlignedDeviceViewTableScan;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.Step.FINAL;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.Step.INTERMEDIATE;
+import static org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode.Step.PARTIAL;
+
+public class TreeViewTest {
+
+ private static final String DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME =
+ String.format("%s.\"%s\"", TREE_VIEW_DB, DEVICE_VIEW_TEST_TABLE);
+
+ // ==================================================================
+ // ===================== Device View Test =======================
+ // ==================================================================
+
+ @Test
+ public void rawDataQueryTest() {
+ PlanTester planTester = new PlanTester();
+
+ LogicalQueryPlan logicalQueryPlan =
+ planTester.createPlan("select * from " + DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME);
+ PlanMatchPattern expectedPlanPattern =
+ output(
+ treeDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("time", "tag1", "tag2", "s1", "s2"),
+ ImmutableSet.of("time", "tag1", "tag2", "s1", "s2")));
+ assertPlan(logicalQueryPlan, expectedPlanPattern);
+
+ // column prune test
+ logicalQueryPlan =
+ planTester.createPlan(
+ "select s1 from " + DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME + " order by tag1");
+ expectedPlanPattern =
+ output(
+ project(
+ streamSort(
+ treeDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")))));
+ assertPlan(logicalQueryPlan, expectedPlanPattern);
+
+ // distributionPlan test
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(
+ project(
+ mergeSort(
+ exchange(),
+ exchange(),
+ treeAlignedDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")),
+ treeNonAlignedDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1"))))));
+
+ assertPlan(
+ planTester.getFragmentPlan(1),
+ treeAlignedDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")));
+
+ assertPlan(
+ planTester.getFragmentPlan(2),
+ treeNonAlignedDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")));
+ }
+
+ @Test
+ public void aggregationQueryTest() {
+ PlanTester planTester = new PlanTester();
+
+ // has non-aligned DeviceEntry, no push-down
+ LogicalQueryPlan logicalQueryPlan =
+ planTester.createPlan(
+ "select tag1, count(s1) from "
+ + DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME
+ + "group by tag1");
+ PlanMatchPattern expectedPlanPattern =
+ output(
+ aggregation(
+ ImmutableMap.of("count", aggregationFunction("count", ImmutableList.of("s1"))),
+ treeDeviceViewTableScan(
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1"))));
+ assertPlan(logicalQueryPlan, expectedPlanPattern);
+
+ // only aligned DeviceEntry, do push-down
+ logicalQueryPlan =
+ planTester.createPlan(
+ "select tag1, count(s1) from "
+ + DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME
+ + " where tag1='shanghai' group by tag1");
+ expectedPlanPattern =
+ output(
+ aggregation(
+ ImmutableMap.of("count", aggregationFunction("count", ImmutableList.of("count_0"))),
+ FINAL,
+ aggregationTreeDeviceViewTableScan(
+ singleGroupingSet("tag1"),
+ ImmutableList.of("tag1"),
+ Optional.empty(),
+ PARTIAL,
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "count_0"),
+ ImmutableSet.of("tag1", "s1"))));
+ assertPlan(logicalQueryPlan, expectedPlanPattern);
+
+ // distributionPlan test
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(
+ aggregation(
+ ImmutableMap.of("count", aggregationFunction("count", ImmutableList.of("count_1"))),
+ FINAL,
+ mergeSort(
+ exchange(),
+ aggregation(
+ ImmutableMap.of(
+ "count_1", aggregationFunction("count", ImmutableList.of("count_0"))),
+ INTERMEDIATE,
+ aggregationTreeDeviceViewTableScan(
+ singleGroupingSet("tag1"),
+ ImmutableList.of("tag1"),
+ Optional.empty(),
+ PARTIAL,
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "count_0"),
+ ImmutableSet.of("tag1", "s1")))))));
+
+ assertPlan(
+ planTester.getFragmentPlan(1),
+ aggregation(
+ ImmutableMap.of("count_1", aggregationFunction("count", ImmutableList.of("count_0"))),
+ INTERMEDIATE,
+ aggregationTreeDeviceViewTableScan(
+ singleGroupingSet("tag1"),
+ ImmutableList.of("tag1"),
+ Optional.empty(),
+ PARTIAL,
+ DEFAULT_TREE_DEVICE_VIEW_TABLE_FULL_NAME,
+ ImmutableList.of("tag1", "count_0"),
+ ImmutableSet.of("tag1", "s1"))));
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanAssert.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanAssert.java
index 4ed4ae4a970..317d8336a9d 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanAssert.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanAssert.java
@@ -80,7 +80,7 @@ public final class PlanAssert {
"Plan does not match, expected [\n\n%s\n] but found [\n\n%s\n], matches:[%s]",
pattern, String.join("\n", actual.accept(new PlanGraphPrinter(), null)), matches));
}
- // TODO support print plan tree
+
PlanNode resolvedPlan = resolveGroupReferences(actual, lookup);
throw new AssertionError(
format(
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
index 57cfde2465e..40e6abba518 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/planner/assertions/PlanMatchPattern.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.Symbol;
import org.apache.iotdb.db.queryengine.plan.relational.planner.iterative.GroupReference;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTableScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.AggregationTreeDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.CollectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.DeviceTableScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.EnforceSingleRowNode;
@@ -41,6 +42,9 @@ import org.apache.iotdb.db.queryengine.plan.relational.planner.node.OutputNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.ProjectNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.SortNode;
import org.apache.iotdb.db.queryengine.plan.relational.planner.node.StreamSortNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeAlignedDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeDeviceViewScanNode;
+import org.apache.iotdb.db.queryengine.plan.relational.planner.node.TreeNonAlignedDeviceViewScanNode;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.DataType;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.SortItem;
@@ -143,6 +147,42 @@ public final class PlanMatchPattern {
dataNodeId));
}
+ public static PlanMatchPattern treeDeviceViewTableScan(
+ String expectedTableName, List<String> outputSymbols, Set<String> assignmentsKeys) {
+ PlanMatchPattern pattern =
+ node(TreeDeviceViewScanNode.class)
+ .with(
+ new DeviceTableScanMatcher(
+ expectedTableName, Optional.empty(), outputSymbols, assignmentsKeys));
+ outputSymbols.forEach(
+ symbol -> pattern.withAlias(symbol, new ColumnReference(expectedTableName, symbol)));
+ return pattern;
+ }
+
+ public static PlanMatchPattern treeAlignedDeviceViewTableScan(
+ String expectedTableName, List<String> outputSymbols, Set<String> assignmentsKeys) {
+ PlanMatchPattern pattern =
+ node(TreeAlignedDeviceViewScanNode.class)
+ .with(
+ new DeviceTableScanMatcher(
+ expectedTableName, Optional.empty(), outputSymbols, assignmentsKeys));
+ outputSymbols.forEach(
+ symbol -> pattern.withAlias(symbol, new ColumnReference(expectedTableName, symbol)));
+ return pattern;
+ }
+
+ public static PlanMatchPattern treeNonAlignedDeviceViewTableScan(
+ String expectedTableName, List<String> outputSymbols, Set<String> assignmentsKeys) {
+ PlanMatchPattern pattern =
+ node(TreeNonAlignedDeviceViewScanNode.class)
+ .with(
+ new DeviceTableScanMatcher(
+ expectedTableName, Optional.empty(), outputSymbols, assignmentsKeys));
+ outputSymbols.forEach(
+ symbol -> pattern.withAlias(symbol, new ColumnReference(expectedTableName, symbol)));
+ return pattern;
+ }
+
public static PlanMatchPattern tableScan(String expectedTableName) {
return node(DeviceTableScanNode.class)
.with(
@@ -288,6 +328,34 @@ public final class PlanMatchPattern {
name, distinct, args, ImmutableList.of(), Optional.empty());
}
+ public static PlanMatchPattern aggregationTreeDeviceViewTableScan(
+ GroupingSetDescriptor groupingSets,
+ List<String> preGroupedSymbols,
+ Optional<Symbol> groupId,
+ AggregationNode.Step step,
+ String expectedTableName,
+ List<String> outputSymbols,
+ Set<String> assignmentsKeys) {
+ PlanMatchPattern result = node(AggregationTreeDeviceViewScanNode.class);
+
+ result.with(
+ new AggregationDeviceTableScanMatcher(
+ groupingSets,
+ preGroupedSymbols,
+ ImmutableList.of(),
+ groupId,
+ step,
+ expectedTableName,
+ Optional.empty(),
+ outputSymbols,
+ assignmentsKeys));
+
+ outputSymbols.forEach(
+ outputSymbol ->
+ result.withAlias(outputSymbol, new ColumnReference(expectedTableName, outputSymbol)));
+ return result;
+ }
+
// Attention: Now we only pass aliases according to outputSymbols, but we don't verify the output
// column if exists in Table and their order because there maybe partial Agg-result.
public static PlanMatchPattern aggregationTableScan(
@@ -393,6 +461,10 @@ public final class PlanMatchPattern {
return builder.build();
}
+ public static PlanMatchPattern streamSort(PlanMatchPattern source) {
+ return node(StreamSortNode.class, source);
+ }
+
public static PlanMatchPattern sort(PlanMatchPattern source) {
return node(SortNode.class, source);
}