You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2024/12/20 11:35:36 UTC
(iotdb) branch TreeToTableView updated: Add FE implement
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch TreeToTableView
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/TreeToTableView by this push:
new 716e8ff9fb4 Add FE implement
716e8ff9fb4 is described below
commit 716e8ff9fb44f4194ba4cf7b25d93f22d259bbcf
Author: Weihao Li <60...@users.noreply.github.com>
AuthorDate: Fri Dec 20 19:35:30 2024 +0800
Add FE implement
---
.../AbstractAggregationTableScanOperator.java | 3 +-
.../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 | 205 +++++++++--
.../iterative/rule/PruneTableScanColumns.java | 52 ++-
.../planner/node/AggregationTableScanNode.java | 387 +++++++++------------
.../node/AggregationTreeDeviceViewScanNode.java | 205 +++++++++++
.../planner/node/DeviceTableScanNode.java | 197 ++++-------
.../node/InformationSchemaTableScanNode.java | 103 +-----
.../relational/planner/node/TableScanNode.java | 118 ++++++-
.../node/TreeAlignedDeviceViewScanNode.java | 126 +++++++
.../planner/node/TreeDeviceViewScanNode.java | 208 +++++++++++
.../node/TreeNonAlignedDeviceViewScanNode.java | 124 +++++++
.../PushAggregationIntoTableScan.java | 4 +
.../optimizations/PushPredicateIntoTableScan.java | 5 +
.../optimizations/UnaliasSymbolReferences.java | 40 ++-
.../relational/planner/optimizations/Util.java | 73 +++-
.../analyzer/MockTableModelDataPartition.java | 8 +-
.../plan/relational/analyzer/TSBSMetadata.java | 23 +-
.../plan/relational/analyzer/TestMatadata.java | 96 ++++-
.../plan/relational/analyzer/TestPlanBuilder.java | 6 +-
.../plan/relational/analyzer/TreeViewTest.java | 180 ++++++++++
.../planner/assertions/PlanMatchPattern.java | 72 ++++
31 files changed, 1914 insertions(+), 558 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggregationTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggregationTableScanOperator.java
index 87db5a3589c..bc264cf83d1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggregationTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/AbstractAggregationTableScanOperator.java
@@ -31,6 +31,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;
@@ -260,7 +261,7 @@ public abstract class AbstractAggregationTableScanOperator extends AbstractDataS
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);
}
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..362b696c97a 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.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;
+ // TODO
+ public String getTreeDBName() {
+ return null;
+ }
- public abstract void setRegionReplicaSet(TRegionReplicaSet regionReplicaSet);
+ public Map<String, String> getMeasurementColumnNameMap() {
+ return null;
+ }
}
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 c6d6fc3a372..91ff285780b 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
@@ -36,6 +36,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;
@@ -338,7 +339,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
@@ -426,7 +427,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 fd533fa9a78..a0224d91f82 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 c392fb357a4..7911b868070 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
@@ -33,6 +33,7 @@ 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;
@@ -41,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;
@@ -58,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;
@@ -485,7 +490,8 @@ public class TableDistributedPlanGenerator
node.getPushDownPredicate(),
node.getPushDownLimit(),
node.getPushDownOffset(),
- node.isPushLimitToEachDevice());
+ node.isPushLimitToEachDevice(),
+ node.containsNonAlignedDevice());
scanNode.setRegionReplicaSet(regionReplicaSet);
return scanNode;
});
@@ -502,15 +508,12 @@ public class TableDistributedPlanGenerator
TRegionReplicaSet mostUsedDataRegion = null;
int maxDeviceEntrySizeOfTableScan = 0;
for (Map.Entry<TRegionReplicaSet, DeviceTableScanNode> entry : tableScanNodeMap.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();
}
}
@@ -524,6 +527,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) {
@@ -603,7 +719,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) {
@@ -644,15 +762,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();
}
}
@@ -696,6 +811,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 =
@@ -703,25 +826,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;
});
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 bd94b448993..2b2a4afb411 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) {
return Optional.of(
new InformationSchemaTableScanNode(
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..d457029078e 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,21 @@ 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);
-
- return new InformationSchemaTableScanNode(
- planNodeId,
- qualifiedObjectName,
- outputSymbols,
- assignments,
- pushDownPredicate,
- pushDownLimit,
- pushDownOffset);
+ 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/TableScanNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TableScanNode.java
index 2fd3764901e..96255c79988 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);
@@ -142,6 +150,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());
@@ -214,6 +226,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..fa9f15670e9
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/planner/node/TreeNonAlignedDeviceViewScanNode.java
@@ -0,0 +1,124 @@
+/*
+ * 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);
+ }
+
+ @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 TreeAlignedDeviceViewScanNode deserialize(ByteBuffer byteBuffer) {
+ TreeAlignedDeviceViewScanNode node = new TreeAlignedDeviceViewScanNode();
+ 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 26dd85498f8..dd6bd86e0c2 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;
@@ -448,6 +449,10 @@ public class PushPredicateIntoTableScan implements PlanOptimizer {
attributeColumns,
queryContext);
tableScanNode.setDeviceEntries(deviceEntries);
+ if (deviceEntries.stream()
+ .anyMatch(deviceEntry -> deviceEntry instanceof NonAlignedAlignedDeviceEntry)) {
+ tableScanNode.setContainsNonAlignedDevice();
+ }
long schemaFetchCost = System.nanoTime() - startTime;
QueryPlanCostMetricSet.getInstance()
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 db6e1addeb0..118e11c6b07 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
@@ -41,6 +41,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.iotdb.db.queryengine.plan.relational.analyzer.TestMatadata.TREE_DB1;
+
public class MockTableModelDataPartition {
private static final SeriesPartitionExecutor EXECUTOR =
@@ -48,8 +50,6 @@ public class MockTableModelDataPartition {
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionExecutorClass(),
IoTDBDescriptor.getInstance().getConfig().getSeriesPartitionSlotNum());
- private static final String DB_NAME = "root.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";
@@ -132,7 +132,7 @@ public class MockTableModelDataPartition {
devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_5), dataRegionMap3);
devicePartitionMap.put(EXECUTOR.getSeriesPartitionSlot(DEVICE_6), dataRegionMap3);
- dbPartitionMap.put(DB_NAME, devicePartitionMap);
+ dbPartitionMap.put(TREE_DB1, devicePartitionMap);
dataPartition.setDataPartitionMap(dbPartitionMap);
return dataPartition;
@@ -161,7 +161,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(TREE_DB1, 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 2747a9e752a..f326106ca9a 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;
@@ -299,40 +300,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 1b49488972e..4a8d727cee3 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.testdb";
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";
+ private static final String TREE_VIEW_DB = "tree_view_db";
+ private static final String DEVICE_VIEW_TEST_TABLE = "root.test.device_view";
+
@Override
public boolean tableExists(QualifiedObjectName name) {
return name.getDatabaseName().equalsIgnoreCase(DB1)
@@ -123,6 +133,29 @@ 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.ID).build(),
+ ColumnSchema.builder(TAG2_CM).setColumnCategory(TsTableColumnCategory.ID).build(),
+ ColumnSchema.builder(S1_CM)
+ .setColumnCategory(TsTableColumnCategory.MEASUREMENT)
+ .build(),
+ ColumnSchema.builder(S2_CM)
+ .setColumnCategory(TsTableColumnCategory.MEASUREMENT)
+ .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(
@@ -242,6 +275,23 @@ public class TestMatadata implements Metadata {
List<Expression> expressionList,
List<String> attributeColumns,
MPPQueryContext context) {
+ if (tableName.getDatabaseName().equals(TREE_VIEW_DB)) {
+ if (expressionList.isEmpty()) {
+ return ImmutableList.of(
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_3.split("\\.")), ImmutableList.of()),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_5.split("\\.")), ImmutableList.of()),
+ new NonAlignedAlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_4.split("\\.")), ImmutableList.of()));
+ }
+
+ return ImmutableList.of(
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_3.split("\\.")), ImmutableList.of()),
+ new AlignedDeviceEntry(
+ new StringArrayDeviceID(DEVICE_5.split("\\.")), ImmutableList.of()));
+ }
if (expressionList.size() == 2) {
if (compareEqualsMatch(expressionList.get(0), "tag1", "beijing")
@@ -249,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(Expression expression, String idOrAttr, String value) {
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..cc926ec0354
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/TreeViewTest.java
@@ -0,0 +1,180 @@
+/*
+ * 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.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 {
+ // ==================================================================
+ // ===================== Device View Test =======================
+ // ==================================================================
+
+ @Test
+ public void rawDataQueryTest() {
+ PlanTester planTester = new PlanTester();
+
+ LogicalQueryPlan logicalQueryPlan =
+ planTester.createPlan("select * from tree_view_db.\"root.test.device_view\"");
+ PlanMatchPattern expectedPlanPattern =
+ output(
+ treeDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ 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 tree_view_db.\"root.test.device_view\" order by tag1");
+ expectedPlanPattern =
+ output(
+ project(
+ streamSort(
+ treeDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")))));
+ assertPlan(logicalQueryPlan, expectedPlanPattern);
+
+ // distributionPlan test
+ assertPlan(
+ planTester.getFragmentPlan(0),
+ output(
+ project(
+ mergeSort(
+ exchange(),
+ treeAlignedDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1")),
+ treeNonAlignedDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ ImmutableList.of("tag1", "s1"),
+ ImmutableSet.of("tag1", "s1"))))));
+
+ assertPlan(
+ planTester.getFragmentPlan(1),
+ treeAlignedDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ 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 tree_view_db.\"root.test.device_view\" group by tag1");
+ PlanMatchPattern expectedPlanPattern =
+ output(
+ aggregation(
+ ImmutableMap.of("count", aggregationFunction("count", ImmutableList.of("s1"))),
+ treeDeviceViewTableScan(
+ "tree_view_db.\"root.test.device_view\"",
+ 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 tree_view_db.\"root.test.device_view\" 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,
+ "tree_view_db.\"root.test.device_view\"",
+ 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,
+ "tree_view_db.\"root.test.device_view\"",
+ 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,
+ "tree_view_db.\"root.test.device_view\"",
+ 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/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);
}